47 lines
1.2 KiB
Python
47 lines
1.2 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.manufacturer import Manufacturer
|
|
|
|
|
|
async def get(db: AsyncSession, manufacturer_id: int) -> Manufacturer | None:
|
|
return await db.get(Manufacturer, manufacturer_id)
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[Manufacturer]:
|
|
result = await db.execute(
|
|
select(Manufacturer).order_by(Manufacturer.name).offset(skip).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
name: str,
|
|
url: str,
|
|
logo_url: str | None = None,
|
|
) -> Manufacturer:
|
|
m = Manufacturer(name=name, url=url, logo_url=logo_url)
|
|
db.add(m)
|
|
await db.commit()
|
|
await db.refresh(m)
|
|
return m
|
|
|
|
|
|
async def update(
|
|
db: AsyncSession, manufacturer: Manufacturer, data: dict
|
|
) -> Manufacturer:
|
|
for field, value in data.items():
|
|
setattr(manufacturer, field, value)
|
|
await db.commit()
|
|
await db.refresh(manufacturer)
|
|
return manufacturer
|
|
|
|
|
|
async def delete(db: AsyncSession, manufacturer: Manufacturer) -> None:
|
|
await db.delete(manufacturer)
|
|
await db.commit()
|