generated_projects/backend/app/crud.py

from sqlalchemy import select, update, delete
from sqlalchemy.ext.asyncio import AsyncSession
from . import models, schemas, database

async def get_item(item_id: int) -> models.Item | None:
    async with database.get_session() as session:
        result = await session.execute(select(models.Item).where(models.Item.id == item_id))
        return result.scalar_one_or_none()

async def get_items(skip: int = 0, limit: int = 100) -> list[models.Item]:
    async with database.get_session() as session:
        result = await session.execute(select(models.Item).offset(skip).limit(limit))
        return result.scalars().all()

async def create_item(item: schemas.ItemCreate) -> models.Item:
    async with database.get_session() as session:
        db_item = models.Item(name=item.name, description=item.description)
        session.add(db_item)
        await session.commit()
        await session.refresh(db_item)
        return db_item

async def update_item(item_id: int, item: schemas.ItemUpdate) -> models.Item | None:
    async with database.get_session() as session:
        stmt = select(models.Item).where(models.Item.id == item_id)
        result = await session.execute(stmt)
        db_item = result.scalar_one_or_none()
        if db_item is None:
            return None
        if item.name is not None:
            db_item.name = item.name
        if item.description is not None:
            db_item.description = item.description
        await session.commit()
        await session.refresh(db_item)
        return db_item

async def delete_item(item_id: int) -> models.Item | None:
    async with database.get_session() as session:
        stmt = select(models.Item).where(models.Item.id == item_id)
        result = await session.execute(stmt)
        db_item = result.scalar_one_or_none()
        if db_item is None:
            return None
        await session.delete(db_item)
        await session.commit()
        return db_item