"""Handle db interactions for platforms.""" from typing import Optional import MySQLdb as mdb from MySQLdb.cursors import Cursor, DictCursor def platforms_by_species( conn: mdb.Connection, speciesid: int, offset: int = 0, limit: Optional[int] = None ) -> tuple[dict, ...]: """Retrieve platforms by the species""" _query = ("SELECT * FROM GeneChip WHERE SpeciesId=%s " "ORDER BY GeneChipName ASC") if bool(limit) and limit > 0:# type: ignore[operator] _query = f"{_query} LIMIT {limit} OFFSET {offset}" with conn.cursor(cursorclass=DictCursor) as cursor: cursor.execute(_query, (speciesid,)) return tuple(dict(row) for row in cursor.fetchall()) def species_platforms_count(conn: mdb.Connection, species_id: int) -> int: """Get the number of platforms in the database for a particular species.""" with conn.cursor(cursorclass=DictCursor) as cursor: cursor.execute( "SELECT COUNT(GeneChipName) AS count FROM GeneChip " "WHERE SpeciesId=%s", (species_id,)) return int(cursor.fetchone()["count"]) def platform_by_id(conn: mdb.Connection, platformid: int) -> Optional[dict]: """Retrieve a platform by its ID""" with conn.cursor(cursorclass=DictCursor) as cursor: cursor.execute("SELECT * FROM GeneChip WHERE Id=%s", (platformid,)) result = cursor.fetchone() if bool(result): return dict(result) return None def platform_by_species_and_id( conn: mdb.Connection, species_id: int, platformid: int ) -> Optional[dict]: """Retrieve a platform by its species and ID""" with conn.cursor(cursorclass=DictCursor) as cursor: cursor.execute("SELECT * FROM GeneChip WHERE SpeciesId=%s AND Id=%s", (species_id, platformid)) result = cursor.fetchone()#pylint: disable=[duplicate-code] if bool(result): return dict(result) return None def save_new_platform(# pylint: disable=[too-many-arguments] cursor: Cursor, species_id: int, geo_platform: str, platform_name: str, platform_shortname: str, platform_title: str, go_tree_value: Optional[str] ) -> dict: """Save a new platform to the database.""" params = { "species_id": species_id, "GeoPlatform": geo_platform, "GeneChipName": platform_name, "Name": platform_shortname, "Title": platform_title, "GO_tree_value": go_tree_value } cursor.execute("SELECT SpeciesId, GeoPlatform FROM GeneChip") assert (species_id, geo_platform) not in ( (row["SpeciesId"], row["GeoPlatform"]) for row in cursor.fetchall()) cursor.execute( "INSERT INTO " "GeneChip(SpeciesId, GeneChipName, Name, GeoPlatform, Title, GO_tree_value) " "VALUES(" "%(species_id)s, %(GeneChipName)s, %(Name)s, %(GeoPlatform)s, " "%(Title)s, %(GO_tree_value)s" ")", params) new_id = cursor.lastrowid cursor.execute("UPDATE GeneChip SET GeneChipId=%s WHERE Id=%s", (new_id, new_id)) return { **params, "Id": new_id, "GeneChipId": new_id }