78 lines
2.3 KiB
Python
78 lines
2.3 KiB
Python
from typing import Annotated
|
|
|
|
import asyncpg
|
|
from dishka.integrations.fastapi import DishkaRoute, FromDishka
|
|
from fastapi import APIRouter, HTTPException, Query
|
|
from pydantic import BaseModel
|
|
|
|
from utils.read import annotations
|
|
from utils.read.models import DEFAULT_LIMIT, AnnotationView, Page
|
|
|
|
router = APIRouter(
|
|
prefix="/api/annotations", tags=["annotations"], route_class=DishkaRoute
|
|
)
|
|
|
|
|
|
class AnnotationCreate(BaseModel):
|
|
account_id: int
|
|
chat_id: int
|
|
message_id: int
|
|
text: str
|
|
|
|
|
|
class AnnotationUpdate(BaseModel):
|
|
text: str
|
|
|
|
|
|
@router.get("")
|
|
async def list_annotations(
|
|
pool: FromDishka[asyncpg.Pool],
|
|
account_id: Annotated[int, Query()],
|
|
chat_id: Annotated[int | None, Query()] = None,
|
|
message_id: Annotated[int | None, Query()] = None,
|
|
limit: Annotated[int, Query()] = DEFAULT_LIMIT,
|
|
offset: Annotated[int, Query()] = 0,
|
|
) -> list[AnnotationView]:
|
|
return await annotations.list_annotations(
|
|
pool,
|
|
account_id,
|
|
Page(limit=limit, offset=offset),
|
|
chat_id=chat_id,
|
|
message_id=message_id,
|
|
)
|
|
|
|
|
|
@router.post("", status_code=201)
|
|
async def create_annotation(
|
|
pool: FromDishka[asyncpg.Pool], body: AnnotationCreate
|
|
) -> AnnotationView:
|
|
return await annotations.create_annotation(
|
|
pool, body.account_id, body.chat_id, body.message_id, body.text
|
|
)
|
|
|
|
|
|
@router.get("/{annotation_id}")
|
|
async def get_annotation(
|
|
pool: FromDishka[asyncpg.Pool], annotation_id: int
|
|
) -> AnnotationView:
|
|
annotation = await annotations.get_annotation(pool, annotation_id)
|
|
if annotation is None:
|
|
raise HTTPException(status_code=404, detail="annotation not found")
|
|
return annotation
|
|
|
|
|
|
@router.put("/{annotation_id}")
|
|
async def update_annotation(
|
|
pool: FromDishka[asyncpg.Pool], annotation_id: int, body: AnnotationUpdate
|
|
) -> AnnotationView:
|
|
annotation = await annotations.update_annotation(pool, annotation_id, body.text)
|
|
if annotation is None:
|
|
raise HTTPException(status_code=404, detail="annotation not found")
|
|
return annotation
|
|
|
|
|
|
@router.delete("/{annotation_id}", status_code=204)
|
|
async def delete_annotation(pool: FromDishka[asyncpg.Pool], annotation_id: int) -> None:
|
|
if not await annotations.delete_annotation(pool, annotation_id):
|
|
raise HTTPException(status_code=404, detail="annotation not found")
|