63 lines
2.1 KiB
Python
63 lines
2.1 KiB
Python
from sqlalchemy.orm import joinedload
|
|
|
|
from modules.clients.models import Client
|
|
from modules.fulfillment_base.models import (
|
|
Marketplace,
|
|
BaseMarketplace,
|
|
)
|
|
from modules.fulfillment_base.schemas.marketplace import (
|
|
CreateMarketplaceSchema,
|
|
UpdateMarketplaceSchema,
|
|
)
|
|
from repositories.mixins import *
|
|
|
|
|
|
class MarketplaceRepository(
|
|
RepCrudMixin[Marketplace, CreateMarketplaceSchema, UpdateMarketplaceSchema],
|
|
):
|
|
session: AsyncSession
|
|
entity_class = Marketplace
|
|
entity_not_found_msg = "Маркетплейс не найден"
|
|
|
|
def _process_get_all_stmt_with_args(self, stmt: Select, *args) -> Select:
|
|
client_id: int = args[0]
|
|
return (
|
|
stmt.options(
|
|
joinedload(Marketplace.base_marketplace),
|
|
joinedload(Marketplace.client),
|
|
)
|
|
.where(
|
|
Marketplace.is_deleted.is_(False), Marketplace.client_id == client_id
|
|
)
|
|
.order_by(Marketplace.id)
|
|
)
|
|
|
|
def _process_get_by_id_stmt(self, stmt: Select) -> Select:
|
|
return stmt.options(
|
|
joinedload(Marketplace.base_marketplace), joinedload(Marketplace.client)
|
|
)
|
|
|
|
async def get_base_marketplaces(self) -> list[BaseMarketplace]:
|
|
stmt = select(BaseMarketplace)
|
|
result = await self.session.execute(stmt)
|
|
return list(result.scalars().all())
|
|
|
|
async def _prepare_create(self, data: CreateMarketplaceSchema) -> dict:
|
|
dict_data = data.model_dump()
|
|
dict_data["base_marketplace_id"] = data.base_marketplace.id
|
|
del dict_data["base_marketplace"]
|
|
dict_data["client_id"] = data.client.id
|
|
del dict_data["client"]
|
|
return dict_data
|
|
|
|
async def update(
|
|
self, template: Marketplace, data: UpdateMarketplaceSchema
|
|
) -> Marketplace:
|
|
if data.base_marketplace:
|
|
data.base_marketplace = BaseMarketplace(
|
|
**data.base_marketplace.model_dump()
|
|
)
|
|
if data.client:
|
|
data.client = Client(**data.client.model_dump())
|
|
return await self._apply_update_data_to_model(template, data, True)
|