70b1c7de10
Add typed service contracts (Protocols + frozen DTOs) in shared/contracts/ for cross-domain communication. Each domain exposes a service interface (BlogService, CalendarService, MarketService, CartService) backed by SQL implementations in shared/services/. A singleton registry with has() guards enables composable startup — apps register their own domain service and stubs for absent domains. Absorbs glue layer: navigation, relationships, event handlers (login, container, order) now live in shared/ with has()-guarded service calls. Factory gains domain_services_fn parameter for per-app service registration. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
58 lines
1.8 KiB
Python
58 lines
1.8 KiB
Python
"""SQL-backed MarketService implementation.
|
|
|
|
Queries ``shared.models.market.*`` and ``shared.models.market_place.*`` —
|
|
only this module may read market-domain tables on behalf of other domains.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from shared.models.market import Product
|
|
from shared.models.market_place import MarketPlace
|
|
from shared.contracts.dtos import MarketPlaceDTO, ProductDTO
|
|
|
|
|
|
def _mp_to_dto(mp: MarketPlace) -> MarketPlaceDTO:
|
|
return MarketPlaceDTO(
|
|
id=mp.id,
|
|
container_type=mp.container_type,
|
|
container_id=mp.container_id,
|
|
name=mp.name,
|
|
slug=mp.slug,
|
|
description=mp.description,
|
|
)
|
|
|
|
|
|
def _product_to_dto(p: Product) -> ProductDTO:
|
|
return ProductDTO(
|
|
id=p.id,
|
|
slug=p.slug,
|
|
title=p.title,
|
|
image=p.image,
|
|
description_short=p.description_short,
|
|
rrp=p.rrp,
|
|
regular_price=p.regular_price,
|
|
special_price=p.special_price,
|
|
)
|
|
|
|
|
|
class SqlMarketService:
|
|
async def marketplaces_for_container(
|
|
self, session: AsyncSession, container_type: str, container_id: int,
|
|
) -> list[MarketPlaceDTO]:
|
|
result = await session.execute(
|
|
select(MarketPlace).where(
|
|
MarketPlace.container_type == container_type,
|
|
MarketPlace.container_id == container_id,
|
|
MarketPlace.deleted_at.is_(None),
|
|
).order_by(MarketPlace.name.asc())
|
|
)
|
|
return [_mp_to_dto(mp) for mp in result.scalars().all()]
|
|
|
|
async def product_by_id(self, session: AsyncSession, product_id: int) -> ProductDTO | None:
|
|
product = (
|
|
await session.execute(select(Product).where(Product.id == product_id))
|
|
).scalar_one_or_none()
|
|
return _product_to_dto(product) if product else None
|