1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
|
"""Database functions for species."""
import math
from typing import Optional
from functools import reduce
import MySQLdb as mdb
from MySQLdb.cursors import DictCursor
def all_species(conn: mdb.Connection) -> tuple:
"Retrieve the species from the database."
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(
"SELECT Id AS SpeciesId, SpeciesName, LOWER(Name) AS Name, "
"MenuName, FullName, TaxonomyId, Family, FamilyOrderId, OrderId "
"FROM Species ORDER BY FamilyOrderId ASC, OrderID ASC")
return tuple(cursor.fetchall())
return tuple()
def order_species_by_family(species: tuple[dict, ...]) -> list:
"""Order the species by their family"""
def __family_order_id__(item):
orderid = item["FamilyOrderId"]
return math.inf if orderid is None else orderid
def __order__(ordered, current):
_key = (__family_order_id__(current), current["Family"])
return {
**ordered,
_key: ordered.get(_key, tuple()) + (current,)
}
ordered = reduce(__order__, species, {})
return sorted(tuple(ordered.items()), key=lambda item: item[0][0])
def species_by_id(conn: mdb.Connection, speciesid) -> dict:
"Retrieve the species from the database by id."
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(
"SELECT Id AS SpeciesId, SpeciesName, LOWER(Name) AS Name, "
"MenuName, FullName, TaxonomyId, Family, FamilyOrderId, OrderId "
"FROM Species WHERE SpeciesId=%s",
(speciesid,))
return cursor.fetchone()
def save_species(conn: mdb.Connection,
common_name: str,
scientific_name: str,
family: str,
taxon_id: Optional[str] = None) -> int:
"""
Save a new species to the database.
Parameters
----------
conn: A connection to the MariaDB database.
taxon_id: The taxonomy identifier for the new species.
common_name: The species' common name.
scientific_name; The species' scientific name.
"""
genus, species = scientific_name.split(" ")
families = species_families(conn)
with conn.cursor() as cursor:
cursor.execute("SELECT MAX(OrderId) FROM Species")
species = {
"common_name": common_name,
"common_name_lower": common_name.lower(),
"menu_name": f"{common_name} ({genus[0]}. {species.lower()})",
"scientific_name": scientific_name,
"family": family,
"family_order": families[family],
"taxon_id": taxon_id,
"species_order": cursor.fetchone()[0] + 5
}
cursor.execute(
"INSERT INTO Species("
"SpeciesName, Name, MenuName, FullName, Family, FamilyOrderId, "
"TaxonomyId, OrderId"
") VALUES ("
"%(common_name)s, %(common_name_lower)s, %(menu_name)s, "
"%(scientific_name)s, %(family)s, %(family_order)s, %(taxon_id)s, "
"%(species_order)s"
")",
species)
species_id = cursor.lastrowid
cursor.execute("UPDATE Species SET SpeciesId=%s WHERE Id=%s",
(species_id, species_id))
return {
**species,
"species_id": species_id
}
def update_species(conn: mdb.Connection,
species_id: int,
common_name: str,
scientific_name: str,
family: str,
family_order: int,
species_order: int):
"""Update a species' details.
Parameters
----------
conn: A connection to the MariaDB database.
species_id: The species identifier
Key-Word Arguments
------------------
common_name: A layman's name for the species
scientific_name: A binomial nomenclature name for the species
family: The grouping under which the species falls
family_order: The ordering for the "family" above
species_order: The ordering of this species in relation to others
"""
with conn.cursor(cursorclass=DictCursor) as cursor:
genus, species = scientific_name.split(" ")
species = {
"species_id": species_id,
"common_name": common_name,
"common_name_lower": common_name.lower(),
"menu_name": f"{common_name} ({genus[0]}. {species.lower()})",
"scientific_name": scientific_name,
"family": family,
"family_order": family_order,
"species_order": species_order
}
cursor.execute(
"UPDATE Species SET "
"SpeciesName=%(common_name)s, "
"Name=%(common_name_lower)s, "
"MenuName=%(menu_name)s, "
"FullName=%(scientific_name)s, "
"Family=%(family)s, "
"FamilyOrderId=%(family_order)s, "
"OrderId=%(species_order)s "
"WHERE Id=%(species_id)s",
species)
def species_families(conn: mdb.Connection) -> dict:
"""Retrieve the families under which species are grouped."""
with conn.cursor(cursorclass=DictCursor) as cursor:
cursor.execute(
"SELECT DISTINCT(Family), FamilyOrderId FROM Species "
"WHERE Family IS NOT NULL")
return {
fam["Family"]: fam["FamilyOrderId"]
for fam in cursor.fetchall()
}
|
