aboutsummaryrefslogtreecommitdiff
path: root/gn3
diff options
context:
space:
mode:
authorFrederick Muriuki Muriithi2023-08-02 10:06:35 +0300
committerFrederick Muriuki Muriithi2023-08-02 10:06:35 +0300
commitabf648858a1a29d452f0efadf53bf5524dde31db (patch)
treeba92d0f5a2f1ef02ba3baac639e69e8c2b4b9864 /gn3
parent1bd01c7396daf9413e9524d7e627e9e36b281213 (diff)
downloadgenenetwork3-abf648858a1a29d452f0efadf53bf5524dde31db.tar.gz
Remove ORM-dependent `fetchall` and `fetchone` functions
Remove Object-Relational Mapping dependent function, `fetchall` and `fetchone` so as to prevent theirs use in the code moving forward.
Diffstat (limited to 'gn3')
-rw-r--r--gn3/db/__init__.py47
1 files changed, 0 insertions, 47 deletions
diff --git a/gn3/db/__init__.py b/gn3/db/__init__.py
index 3b063f1..5676eb5 100644
--- a/gn3/db/__init__.py
+++ b/gn3/db/__init__.py
@@ -63,53 +63,6 @@ def update(conn: Any,
conn.commit()
return cursor.rowcount
-
-def fetchone(conn: Any,
- table: str,
- where: Optional[Dataclass],
- columns: Union[str, List[str]] = "*") -> Optional[Dataclass]:
- """Run a SELECT on a table. Returns only one result!"""
- if not any(astuple(where)):
- return None
- where_ = {TABLEMAP[table].get(k): v for k, v in asdict(where).items()
- if v is not None and k in TABLEMAP[table]}
- sql = ""
- if columns != "*":
- sql = f"SELECT {', '.join(columns)} FROM {table} "
- else:
- sql = f"SELECT * FROM {table} "
- if where:
- sql += "WHERE "
- sql += " AND ".join(f"{k} = "
- "%s" for k in where_.keys())
- with conn.cursor() as cursor:
- cursor.execute(sql, tuple(where_.values()))
- return DATACLASSMAP[table](*cursor.fetchone())
-
-
-def fetchall(conn: Any,
- table: str,
- where: Optional[Dataclass],
- columns: Union[str, List[str]] = "*") -> Optional[Generator]:
- """Run a SELECT on a table. Returns all the results as a tuple!"""
- if not any(astuple(where)):
- return None
- where_ = {TABLEMAP[table].get(k): v for k, v in asdict(where).items()
- if v is not None and k in TABLEMAP[table]}
- sql = ""
- if columns != "*":
- sql = f"SELECT {', '.join(columns)} FROM {table} "
- else:
- sql = f"SELECT * FROM {table} "
- if where:
- sql += "WHERE "
- sql += " AND ".join(f"{k} = "
- "%s" for k in where_.keys())
- with conn.cursor() as cursor:
- cursor.execute(sql, tuple(where_.values()))
- return (DATACLASSMAP[table](*record) for record in cursor.fetchall())
-
-
def insert(conn: Any,
table: str,
data: Dataclass) -> Optional[int]: