a

Backend/utils/db.py

@@ -0,0 +1,53 @@
+import aiosqlite
+from typing import Any, List, Tuple, Optional, Dict, Union
+
+DB_PATH = "database.sqlite3"
+
+# Generated by Github Copilot
+# db 사용을 편하게 하기 위한 함수
+
+
+async def get_db_connection(db_path: str = DB_PATH) -> aiosqlite.Connection:
+    return await aiosqlite.connect(db_path)
+
+
+async def execute(
+    query: str,
+    params: Union[Tuple[Any, ...], Dict[str, Any]] = (),
+    db_path: str = DB_PATH,
+) -> None:
+    async with aiosqlite.connect(db_path) as db:
+        await db.execute(query, params)
+        await db.commit()
+
+
+async def fetch_one(
+    query: str,
+    params: Union[Tuple[Any, ...], Dict[str, Any]] = (),
+    db_path: str = DB_PATH,
+) -> Optional[aiosqlite.Row]:
+    async with aiosqlite.connect(db_path) as db:
+        db.row_factory = aiosqlite.Row
+        async with db.execute(query, params) as cursor:
+            return await cursor.fetchone()
+
+
+async def fetch_all(
+    query: str,
+    params: Union[Tuple[Any, ...], Dict[str, Any]] = (),
+    db_path: str = DB_PATH,
+) -> List[aiosqlite.Row]:
+    async with aiosqlite.connect(db_path) as db:
+        db.row_factory = aiosqlite.Row
+        async with db.execute(query, params) as cursor:
+            return await cursor.fetchall()
+
+
+async def executemany(
+    query: str,
+    seq_of_params: List[Union[Tuple[Any, ...], Dict[str, Any]]],
+    db_path: str = DB_PATH,
+) -> None:
+    async with aiosqlite.connect(db_path) as db:
+        await db.executemany(query, seq_of_params)
+        await db.commit()