73 lines
2.2 KiB
Python
73 lines
2.2 KiB
Python
import asyncpg
|
|
|
|
from utils.read.models import AnnotationView, Page
|
|
|
|
_COLS = "id, account_id, chat_id, message_id, text, created_at, updated_at"
|
|
|
|
|
|
async def list_annotations(
|
|
pool: asyncpg.Pool,
|
|
account_id: int,
|
|
page: Page,
|
|
*,
|
|
chat_id: int | None = None,
|
|
message_id: int | None = None,
|
|
) -> list[AnnotationView]:
|
|
params: list[object] = [account_id]
|
|
where = "account_id = $1"
|
|
if chat_id is not None:
|
|
params.append(chat_id)
|
|
where += f" AND chat_id = ${len(params)}"
|
|
if message_id is not None:
|
|
params.append(message_id)
|
|
where += f" AND message_id = ${len(params)}"
|
|
params.append(page.capped_limit)
|
|
params.append(page.offset)
|
|
rows = await pool.fetch(
|
|
f"SELECT {_COLS} FROM annotations WHERE {where} " # noqa: S608
|
|
f"ORDER BY created_at DESC LIMIT ${len(params) - 1} OFFSET ${len(params)}",
|
|
*params,
|
|
)
|
|
return [AnnotationView(**dict(row)) for row in rows]
|
|
|
|
|
|
async def get_annotation(
|
|
pool: asyncpg.Pool, annotation_id: int
|
|
) -> AnnotationView | None:
|
|
row = await pool.fetchrow(
|
|
f"SELECT {_COLS} FROM annotations WHERE id = $1", # noqa: S608
|
|
annotation_id,
|
|
)
|
|
return AnnotationView(**dict(row)) if row else None
|
|
|
|
|
|
async def create_annotation(
|
|
pool: asyncpg.Pool, account_id: int, chat_id: int, message_id: int, text: str
|
|
) -> AnnotationView:
|
|
row = await pool.fetchrow(
|
|
"INSERT INTO annotations (account_id, chat_id, message_id, text) " # noqa: S608
|
|
f"VALUES ($1, $2, $3, $4) RETURNING {_COLS}",
|
|
account_id,
|
|
chat_id,
|
|
message_id,
|
|
text,
|
|
)
|
|
return AnnotationView(**dict(row))
|
|
|
|
|
|
async def update_annotation(
|
|
pool: asyncpg.Pool, annotation_id: int, text: str
|
|
) -> AnnotationView | None:
|
|
row = await pool.fetchrow(
|
|
"UPDATE annotations SET text = $2, updated_at = now() " # noqa: S608
|
|
f"WHERE id = $1 RETURNING {_COLS}",
|
|
annotation_id,
|
|
text,
|
|
)
|
|
return AnnotationView(**dict(row)) if row else None
|
|
|
|
|
|
async def delete_annotation(pool: asyncpg.Pool, annotation_id: int) -> bool:
|
|
result = await pool.execute("DELETE FROM annotations WHERE id = $1", annotation_id)
|
|
return result.endswith("1")
|