48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.mariadb.device import Device, DeviceGroup
|
|
from app.repositories.base import BaseRepository
|
|
|
|
|
|
class DeviceRepository(BaseRepository[Device]):
|
|
def __init__(self, session: AsyncSession):
|
|
super().__init__(Device, session)
|
|
|
|
async def get_by_uid(self, device_uid: str) -> Device | None:
|
|
stmt = select(Device).where(Device.device_uid == device_uid, Device.is_deleted == False) # noqa: E712
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_by_owner(self, owner_id: int, skip: int = 0, limit: int = 100) -> list[Device]:
|
|
stmt = (
|
|
select(Device)
|
|
.where(Device.owner_id == owner_id, Device.is_deleted == False) # noqa: E712
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def get_by_group(self, group_id: int, skip: int = 0, limit: int = 100) -> list[Device]:
|
|
stmt = (
|
|
select(Device)
|
|
.where(Device.group_id == group_id, Device.is_deleted == False) # noqa: E712
|
|
.offset(skip)
|
|
.limit(limit)
|
|
)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
|
|
class DeviceGroupRepository(BaseRepository[DeviceGroup]):
|
|
def __init__(self, session: AsyncSession):
|
|
super().__init__(DeviceGroup, session)
|
|
|
|
async def get_by_name(self, name: str) -> DeviceGroup | None:
|
|
stmt = select(DeviceGroup).where(DeviceGroup.name == name)
|
|
result = await self.session.execute(stmt)
|
|
return result.scalar_one_or_none()
|