40 lines
1.1 KiB
Python
40 lines
1.1 KiB
Python
from typing import Any
|
|
from database.connection import get_db
|
|
|
|
|
|
async def get_all_organizations() -> list[dict]:
|
|
db = get_db()
|
|
async with db.execute("SELECT * FROM organizations ORDER BY name") as cur:
|
|
rows = await cur.fetchall()
|
|
return [dict(r) for r in rows]
|
|
|
|
|
|
async def get_organization(org_id: int) -> dict | None:
|
|
db = get_db()
|
|
async with db.execute("SELECT * FROM organizations WHERE id = ?", (org_id,)) as cur:
|
|
row = await cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
|
|
async def create_organization(name: str) -> int:
|
|
db = get_db()
|
|
async with db.execute(
|
|
"INSERT INTO organizations (name) VALUES (?)", (name,)
|
|
) as cur:
|
|
org_id = cur.lastrowid
|
|
await db.commit()
|
|
return org_id
|
|
|
|
|
|
async def update_organization(org_id: int, name: str) -> None:
|
|
db = get_db()
|
|
await db.execute(
|
|
"UPDATE organizations SET name = ? WHERE id = ?", (name, org_id)
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
async def delete_organization(org_id: int) -> None:
|
|
db = get_db()
|
|
await db.execute("DELETE FROM organizations WHERE id = ?", (org_id,))
|
|
await db.commit()
|