57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.scope_model import ScopeModel
|
|
|
|
|
|
async def get(db: AsyncSession, scope_model_id: int) -> ScopeModel | None:
|
|
return await db.get(ScopeModel, scope_model_id)
|
|
|
|
|
|
async def get_by_name_and_manufacturer(
|
|
db: AsyncSession, name: str, manufacturer_id: int
|
|
) -> ScopeModel | None:
|
|
result = await db.execute(
|
|
select(ScopeModel).where(
|
|
ScopeModel.name == name,
|
|
ScopeModel.manufacturer_id == manufacturer_id,
|
|
)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
|
|
async def list_all(
|
|
db: AsyncSession, skip: int = 0, limit: int = 100
|
|
) -> list[ScopeModel]:
|
|
result = await db.execute(
|
|
select(ScopeModel).order_by(ScopeModel.name).offset(skip).limit(limit)
|
|
)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
async def create(
|
|
db: AsyncSession,
|
|
*,
|
|
name: str,
|
|
manufacturer_id: int,
|
|
url: str | None = None,
|
|
) -> ScopeModel:
|
|
sm = ScopeModel(name=name, manufacturer_id=manufacturer_id, url=url)
|
|
db.add(sm)
|
|
await db.commit()
|
|
await db.refresh(sm, attribute_names=["manufacturer"])
|
|
return sm
|
|
|
|
|
|
async def update(db: AsyncSession, scope_model: ScopeModel, data: dict) -> ScopeModel:
|
|
for field, value in data.items():
|
|
setattr(scope_model, field, value)
|
|
await db.commit()
|
|
await db.refresh(scope_model, attribute_names=["manufacturer"])
|
|
return scope_model
|
|
|
|
|
|
async def delete(db: AsyncSession, scope_model: ScopeModel) -> None:
|
|
await db.delete(scope_model)
|
|
await db.commit()
|