92 lines
3.1 KiB
Python
92 lines
3.1 KiB
Python
from typing import Optional, List, Tuple
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
from sqlalchemy.orm import selectinload
|
|
|
|
from app.models.medicine import Medicine
|
|
|
|
|
|
class MedicineRepository:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
|
|
async def get_by_id(self, medicine_id: int) -> Optional[Medicine]:
|
|
result = await self.db.execute(
|
|
select(Medicine)
|
|
.options(selectinload(Medicine.batches))
|
|
.where(Medicine.id == medicine_id)
|
|
)
|
|
return result.scalar_one_or_none()
|
|
|
|
async def get_list(
|
|
self,
|
|
category_id: Optional[int] = None,
|
|
search: Optional[str] = None,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> Tuple[List[Medicine], int]:
|
|
query = select(Medicine).options(selectinload(Medicine.batches))
|
|
|
|
if category_id:
|
|
query = query.where(Medicine.category_id == category_id)
|
|
|
|
if search:
|
|
search_filter = f"%{search}%"
|
|
query = query.where(
|
|
Medicine.name.ilike(search_filter) |
|
|
Medicine.generic_name.ilike(search_filter) |
|
|
Medicine.brand_name.ilike(search_filter)
|
|
)
|
|
|
|
count_query = select(func.count()).select_from(Medicine)
|
|
if category_id:
|
|
count_query = count_query.where(Medicine.category_id == category_id)
|
|
if search:
|
|
search_filter = f"%{search}%"
|
|
count_query = count_query.where(
|
|
Medicine.name.ilike(search_filter) |
|
|
Medicine.generic_name.ilike(search_filter) |
|
|
Medicine.brand_name.ilike(search_filter)
|
|
)
|
|
|
|
total_result = await self.db.execute(count_query)
|
|
total = total_result.scalar()
|
|
|
|
query = query.offset((page - 1) * page_size).limit(page_size)
|
|
result = await self.db.execute(query)
|
|
medicines = list(result.scalars().all())
|
|
|
|
return medicines, total
|
|
|
|
async def create(self, data: dict) -> Medicine:
|
|
medicine = Medicine(**data)
|
|
self.db.add(medicine)
|
|
await self.db.flush()
|
|
return medicine
|
|
|
|
async def update(self, medicine_id: int, data: dict) -> Optional[Medicine]:
|
|
medicine = await self.get_by_id(medicine_id)
|
|
if not medicine:
|
|
return None
|
|
for key, value in data.items():
|
|
setattr(medicine, key, value)
|
|
await self.db.flush()
|
|
return medicine
|
|
|
|
async def delete(self, medicine_id: int) -> bool:
|
|
medicine = await self.get_by_id(medicine_id)
|
|
if not medicine:
|
|
return False
|
|
await self.db.delete(medicine)
|
|
return True
|
|
|
|
async def search(self, query: str) -> List[Medicine]:
|
|
search_filter = f"%{query}%"
|
|
result = await self.db.execute(
|
|
select(Medicine).options(selectinload(Medicine.batches)).where(
|
|
Medicine.name.ilike(search_filter) |
|
|
Medicine.generic_name.ilike(search_filter) |
|
|
Medicine.indications.ilike(search_filter)
|
|
)
|
|
)
|
|
return list(result.scalars().all()) |