46 lines
No EOL
1,000 B
Python
46 lines
No EOL
1,000 B
Python
from typing import Sequence
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy import select, delete
|
|
from models.items import Item
|
|
from schemas.items import ItemIn
|
|
|
|
|
|
async def get_item_by_id(id: int, db: AsyncSession) -> Item | None:
|
|
result = await db.execute(select(Item).where(Item.id == id))
|
|
|
|
item = result.scalar_one_or_none()
|
|
|
|
return item
|
|
|
|
|
|
async def get_items(db: AsyncSession) -> Sequence[Item]:
|
|
result = await db.execute(select(Item))
|
|
|
|
items = result.scalars().all()
|
|
|
|
return items
|
|
|
|
|
|
async def create_item(payload: ItemIn, db: AsyncSession) -> Item:
|
|
item = Item(text=payload.text)
|
|
|
|
db.add(item)
|
|
await db.commit()
|
|
await db.refresh(item)
|
|
|
|
return item
|
|
|
|
|
|
async def update_item(item: Item, payload: ItemIn, db: AsyncSession) -> Item:
|
|
item.text = payload.text
|
|
await db.commit()
|
|
|
|
await db.refresh(item)
|
|
|
|
return item
|
|
|
|
|
|
async def delete_item(item: Item, db: AsyncSession):
|
|
|
|
await db.delete(item)
|
|
await db.commit() |