64 lines
2.3 KiB
Python
64 lines
2.3 KiB
Python
from typing import List, Optional, Tuple
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.medicine import Medicine
|
|
from app.schemas.medicine import MedicineCreate, MedicineUpdate, MedicineWithStock
|
|
from app.repositories.medicine import MedicineRepository
|
|
|
|
|
|
class MedicineService:
|
|
def __init__(self, db: AsyncSession):
|
|
self.db = db
|
|
self.repo = MedicineRepository(db)
|
|
|
|
async def get_medicines(
|
|
self,
|
|
category_id: Optional[int] = None,
|
|
search: Optional[str] = None,
|
|
page: int = 1,
|
|
page_size: int = 20
|
|
) -> Tuple[List[MedicineWithStock], int]:
|
|
medicines, total = await self.repo.get_list(
|
|
category_id=category_id,
|
|
search=search,
|
|
page=page,
|
|
page_size=page_size
|
|
)
|
|
|
|
result = []
|
|
for medicine in medicines:
|
|
total_quantity = sum(b.quantity for b in medicine.batches if not b.is_expired)
|
|
|
|
nearest_expiry = None
|
|
for batch in medicine.batches:
|
|
if not batch.is_expired:
|
|
if nearest_expiry is None or batch.expiry_date < nearest_expiry:
|
|
nearest_expiry = batch.expiry_date
|
|
|
|
medicine_with_stock = MedicineWithStock(
|
|
**medicine.__dict__,
|
|
total_quantity=total_quantity,
|
|
nearest_expiry_date=nearest_expiry,
|
|
batch_count=len([b for b in medicine.batches if not b.is_expired])
|
|
)
|
|
result.append(medicine_with_stock)
|
|
|
|
return result, total
|
|
|
|
async def get_medicine(self, medicine_id: int) -> Optional[Medicine]:
|
|
return await self.repo.get_by_id(medicine_id)
|
|
|
|
async def create_medicine(self, data: MedicineCreate, user_id: int) -> Medicine:
|
|
medicine_data = data.model_dump()
|
|
medicine_data['created_by'] = user_id
|
|
return await self.repo.create(medicine_data)
|
|
|
|
async def update_medicine(self, medicine_id: int, data: MedicineUpdate) -> Optional[Medicine]:
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
return await self.repo.update(medicine_id, update_data)
|
|
|
|
async def delete_medicine(self, medicine_id: int) -> bool:
|
|
return await self.repo.delete(medicine_id)
|
|
|
|
async def search_medicines(self, query: str) -> List[Medicine]:
|
|
return await self.repo.search(query) |