53 lines
1.5 KiB
Python
53 lines
1.5 KiB
Python
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()
|