84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
from typing import TYPE_CHECKING, Optional
|
|
|
|
from litestar import Controller, get, post
|
|
from litestar.contrib.repository.filters import LimitOffset, SearchFilter
|
|
from litestar.di import Provide
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.domain.company import Company
|
|
from app.domain.asset_item import (
|
|
AssetItem,
|
|
AssetItemReadDTO,
|
|
AssetItemWriteDTO,
|
|
Repository,
|
|
Service,
|
|
)
|
|
from app.lib.responses import ObjectListResponse, ObjectResponse
|
|
|
|
if TYPE_CHECKING:
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
|
|
DETAIL_ROUTE = "/{asset_item_id:int}"
|
|
|
|
|
|
async def provides_service(db_session: AsyncSession, company_id: int) -> Service:
|
|
"""Constructs repository and service objects for the request."""
|
|
from app.controllers.company import provides_service
|
|
|
|
company_service = provides_service(db_session)
|
|
company = await company_service.get(company_id)
|
|
return Service(Repository(session=db_session, company=company))
|
|
|
|
|
|
async def get_company(db_session: AsyncSession, company_id: int) -> Company:
|
|
from app.controllers.company import provides_service
|
|
|
|
company_service = provides_service(db_session)
|
|
return await company_service.get(company_id)
|
|
|
|
|
|
class AssetItemController(Controller):
|
|
dto = AssetItemWriteDTO
|
|
return_dto = AssetItemReadDTO
|
|
path = "/company/{company_id:int}/asset-items"
|
|
dependencies = {
|
|
"service": Provide(provides_service, sync_to_thread=False),
|
|
}
|
|
tags = ["AssetItems"]
|
|
|
|
@post()
|
|
async def create_asset_item(
|
|
self, data: AssetItem, service: Service
|
|
) -> AssetItem:
|
|
return await service.create(data)
|
|
|
|
@get()
|
|
async def get_asset_items(
|
|
self,
|
|
service: Service,
|
|
search: Optional[str] = None,
|
|
) -> ObjectListResponse[AssetItem]:
|
|
filters = [
|
|
LimitOffset(limit=20, offset=0),
|
|
]
|
|
|
|
if search is not None:
|
|
filters.append(
|
|
SearchFilter(
|
|
field_name="caption",
|
|
value=search,
|
|
),
|
|
)
|
|
|
|
content = await service.list(*filters)
|
|
return ObjectListResponse(content=content)
|
|
|
|
@get(DETAIL_ROUTE)
|
|
async def get_asset_item(
|
|
self, service: Service, asset_item_id: int
|
|
) -> ObjectResponse[AssetItem]:
|
|
content = await service.get(asset_item_id)
|
|
return ObjectResponse(content=content)
|
|
|