38 lines
1.3 KiB
Python
38 lines
1.3 KiB
Python
from typing import Union
|
|
|
|
from sqlalchemy import select
|
|
|
|
import models
|
|
from schemas.shipping_warehouse import GetAllShippingWarehousesResponse, ShippingWarehouseSchema
|
|
from services.base import BaseService
|
|
|
|
|
|
class ShippingWarehouseService(BaseService):
|
|
async def get_all(self) -> GetAllShippingWarehousesResponse:
|
|
stmt = (
|
|
select(
|
|
models.ShippingWarehouse
|
|
)
|
|
.order_by(
|
|
models.ShippingWarehouse.id
|
|
)
|
|
)
|
|
shipping_warehouses = (await self.session.scalars(stmt)).all()
|
|
result = []
|
|
for shipping_warehouse in shipping_warehouses:
|
|
result.append(ShippingWarehouseSchema.model_validate(shipping_warehouse))
|
|
return GetAllShippingWarehousesResponse(shipping_warehouses=result)
|
|
|
|
async def get_by_name(self, name: str) -> Union[models.ShippingWarehouse, None]:
|
|
stmt = select(models.ShippingWarehouse).where(models.ShippingWarehouse.name == name)
|
|
shipping_warehouse = await self.session.scalar(stmt)
|
|
return shipping_warehouse
|
|
|
|
async def create_by_name(self, name: str) -> models.ShippingWarehouse:
|
|
shipping_warehouse = models.ShippingWarehouse(
|
|
name=name
|
|
)
|
|
self.session.add(shipping_warehouse)
|
|
await self.session.flush()
|
|
return shipping_warehouse
|