Skip to content

Services

Work in progress...

This page is a work in progress, please check back later for updates.

AI generated content

This page was generated by an AI model trained on the original content and may not be fully accurate.

Services are one of the core building blocks of the Plateforme framework. They encapsulate your application's business logic and operations while providing a clean interface for interacting with resources and managing database transactions.

Overview

A service in Plateforme is a Python class that defines:

  • Business logic and operations for resources
  • Transaction management
  • API endpoint implementations
  • Validation and authorization rules

Key concepts

Service definition

Services are defined by inheriting from BaseService or BaseServiceWithSpec:

from plateforme import BaseServiceWithSpec
from plateforme.api import AsyncSessionDep, route

from .resources import User

class UserService(BaseServiceWithSpec[User]):
    @route.post('/login')
    async def authenticate(
        self, session: AsyncSessionDep, username: str, password: str
    ) -> bool:
        ...
    
    @route.post()
    async def register(
        self, session: AsyncSessionDep, payload: User.Register
    ) -> User:
        ...

Service configuration

Services can be configured through their inner __config__ attribute. It can be provided as a dictionary or a ServiceConfig object.

The ConfigDict class provides a convenient way to define configuration options, but you can also use a raw dictionary. Type hints should work with both approaches.

from plateforme import BaseService, ConfigDict

class UserService(BaseService):
    __config__ = ConfigDict(
        name="user_management",
        limit=100,
    )
    ...

CRUD operations

The CRUDService class provides common database operations out of the box and can be assigned to a resource:

from plateforme import BaseResource, ConfigDict, CRUDService


class Product(BaseResource):
    __config__ = ConfigDict(services=(CRUDService(include_method=['GET']),))

Service specifications

Services can use specifications to define model schemas and validation rules:

from plateforme import BaseServiceWithSpec, CRUDSpec, route

class OrderSpec(CRUDSpec):
    class Create:
        user_id: int
        items: list[str]
    
    class Read:
        id: int
        user_id: int
        items: list[str]
        total: float

class OrderService(BaseServiceWithSpec[OrderSpec]):
    @route.post()
    async def create_order(self, data: OrderSpec.Create) -> OrderSpec.Read:
        ...

The framework also provides powerful capabilities to dynamically generate schema models using the __apply__ method and field lookup configuration:

from plateforme import BaseSpec, SpecFacade

class CustomSpec(BaseSpec):
    @classmethod
    def __apply__(cls, facade: SpecFacade) -> None:
        # Register a schema by collecting fields based on lookup filters
        facade._register_schema(
            'CustomSchema',
            __collect__=(
                {
                    'exclude': {'frozen': True},   # Exclude frozen fields
                    'partial': True,               # Make fields optional
                    'override': {                  # Override field attributes
                        'include': {'linked': True},
                        'update': {'target_ref': True},
                    },
                },
            ),
        )

This approach allows you to dynamically generate schema models by applying lookup filters against a base model, enabling flexible schema definitions without manually defining each field. The __collect__ parameter accepts a tuple of lookup configurations that specify which fields to include, exclude, or modify from the resource model.

Advanced features

Selection context

Selection provides type-safe, query-building mechanisms to retrieve resources by their identifiers or specific criteria. It handles database query construction with proper join relationships and filtering. This abstraction ensures consistent resource resolution throughout the application while supporting both individual key lookups and list-based selections. Selection objects can be directly used as dependencies in route handlers.

from plateforme import BaseServiceWithSpec, BaseSpec
from plateforme.api import AsyncSessionDep, Body, KeyList, Selection, Query, route

class CustomSpec(BaseSpec):
    class Schema:
        ...

class CustomService(BaseServiceWithSpec[CustomSpec]):
    @route.get()
    async def read_many(
        self,
        session: AsyncSessionDep,
        selection: KeyList[CustomSpec] = Selection(),
        filter: Filter | None = Body(default=None),
        sort: Sort | None = Query(default=None),
    ) -> list[CustomSpec.Schema]:
        query = selection.build_query(
            filter_criteria=filter,
            sort_criteria=sort,
        )
        buffer = await session.execute(query)
        result = buffer.unique().scalars().all()
        ...

Bulk operations

Bulk operations optimize database performance by managing multiple related entities in a single transaction. The bulk context tracks entity references and dependencies, handling resolution automatically. This reduces database roundtrips and ensures data integrity across complex object graphs. Bulk contexts support different resolution strategies like 'hydrate' for eager loading or 'bind' for more efficient operations when full object hydration isn't needed.

from plateforme import BaseServiceWithSpec, BaseSpec
from plateforme.api import AsyncSessionDep, Payload, route

class CustomSpec(BaseSpec):
    class Schema:
        ...

class CustomService(BaseServiceWithSpec[CustomSpec]):
    @route.post()
    async def create(
        self,
        session: AsyncSessionDep,
        payload: CustomSpec.Schema = Payload(),
    ) -> Any:
        async with session.bulk() as bulk:
            result = self.resource_adapter.validate_python(payload)
            await bulk.resolve(raise_errors=True, scope='references')
            session.add_all(result)
        await session.commit()
        ...

Payload handling

Payload objects validate incoming data against resource specifications and transform it into proper model instances. They enforce type safety, field constraints, and business rules. The apply_selection parameter determines whether selection context should influence payload validation. Payloads support both individual objects and collections, with appropriate schema models defined in specifications. They integrate directly with route parameters for automatic request validation.

from plateforme import BaseServiceWithSpec, BaseSpec
from plateforme.api import AsyncSessionDep, Key, Selection, Payload, route

class CustomSpec(BaseSpec):
    class Schema:
        ...

class CustomService(BaseServiceWithSpec[CustomSpec]):
    @route.put()
    async def update_one(
        self,
        session: AsyncSessionDep,
        selection: Key[CustomSpec] = Selection(),
        payload: CustomSpec.Schema = Payload(apply_selection=False),
    ) -> CustomSpec.Schema:
        # Usage
        update = payload.model_dump(mode='raw', exclude_unset=True)
        result_source = await selection.resolve(session)
        result_dirty = {**result_source.resource_dump(mode='raw'), **update}
        ...

Filter and sort capabilities

Filter and Sort objects provide rich query criteria that map directly to database operations. Filters support complex conditions with operators (eq, gt, lt, like, etc.), logical combinations (and, or), and nested field paths. They can be provided in multiple formats, including query string notation and structured JSON. Sort objects define ordering with field paths, direction modifiers, and null handling. Both are designed to be user-friendly while preventing SQL injection and ensuring optimal query performance.

# Parameter definitions
filter: Filter | None = Body(
    default=None,
    examples=[
        '.user.name=like~Al*',
        {'price': {'operator': 'gt', 'value': 1000}}
    ]
)
sort: Sort | None = Query(
    default=None,
    examples=['user.name,-price']
)

# Usage
query = selection.build_query(
    filter_criteria=filter,
    sort_criteria=sort,
)