Microservice Pattern

This guide describes how to create and maintain Python microservice components in the Control Plane. Microservices provide domain-specific CRUD operations and business logic against a database. They are designed to be included in larger deployment units like the Monolith.

When to Use This Pattern

Use the microservice pattern when:

You need domain-specific CRUD operations against a database
The functionality requires business logic validation
The service needs to be reusable across different deployment targets

Do not use this pattern for:

Web app UIs (use a dedicated frontend component)
Scripts or one-off utilities
Shared library code (use a utility package in components/ without the full service structure)

Directory Structure

Every microservice follows this structure:

components/{service_name}/
├── BUILD                          # Pants build configuration (root)
├── requirements.txt               # Python dependencies
└── {service_name}/
    ├── BUILD                      # Module-level build config
    ├── index.py                   # Service implementation and binding function
    ├── config.py                  # Configuration with cloud variants
    ├── conftest.py                # pytest fixtures
    ├── data/                      # Data access layer (optional)
    │   ├── BUILD
    │   └── {name}_data_service.py
    ├── testing/
    │   ├── BUILD
    │   └── fixtures.py            # Test fixtures for consumers
    └── tests/
        ├── BUILD
        └── *_test.py

Step-by-Step Guide

1. Create the API Definition

First, create the API contract in components/apis/{service_name}_api/:

# components/apis/{service_name}_api/__init__.py
import abc
from pydantic import BaseModel


class CreateThingRequest(BaseModel):
    name: str
    description: str


class CreateThingResponse(BaseModel):
    thing_id: str


class GetThingRequest(BaseModel):
    thing_id: str


class GetThingResponse(BaseModel):
    thing_id: str
    name: str
    description: str


class ThingService(abc.ABC):
    @abc.abstractmethod
    def create_thing(self, request: CreateThingRequest) -> CreateThingResponse:
        """Creates a new thing."""
        ...

    @abc.abstractmethod
    def get_thing(self, request: GetThingRequest) -> GetThingResponse:
        """Gets a thing by ID."""
        ...

Create the BUILD file:

# components/apis/{service_name}_api/BUILD
python_sources()

2. Create the Service Component

Root BUILD File

The default resolve is parametrized to include the microservice's own name, as well as any other services that need to compose it into a larger deployment unit (e.g. BFFs and Monolith)

# components/{service_name}/BUILD
__defaults__(
    {
        ("python_sources", "python_requirements"): dict(
            resolve=parametrize("monolith", "{service_name}"),
        ),
    },
    all=dict(resolve="{service_name}"),
)

python_requirements(
    name="reqs",
)

requirements.txt

# components/{service_name}/requirements.txt
# Add any service-specific dependencies here

Module BUILD File

# components/{service_name}/{service_name}/BUILD
python_sources()

python_test_utils(
    name="test_utils",
    dependencies=[
        "//components/service_utils/service_utils/testing/fixtures.py",
        "//components/{service_name}/{service_name}/testing/fixtures.py",
    ],
)

3. Create the Configuration

# components/{service_name}/{service_name}/config.py
from service_utils.config import AWSBaseConfig, BaseConfig, GCPBaseConfig


class Config(BaseConfig):
    # Add service-specific, cloud-agnostic configuration fields here
    example_table_name: str


# Required: Used for AWS deployment
class AWSConfig(Config, AWSBaseConfig):
    # AWS-specific overrides (usually empty)
    ...


# Optional: GCP is not currently supported
class GCPConfig(Config, GCPBaseConfig):
    # GCP-specific overrides (usually empty)
    ...

4. Create the Data Service (Optional)

If your service needs database access, create an abstract data service with cloud-specific implementations:

# components/{service_name}/{service_name}/data/{name}_data_service.py
import abc
from injector import Binder, inject, singleton
from pydantic import BaseModel


class ThingRecord(BaseModel):
    """Database record model."""
    thing_id: str
    name: str
    description: str


class ThingDataService(abc.ABC):
    """Abstract data access layer."""

    @abc.abstractmethod
    def create(self, record: ThingRecord) -> None:
        ...

    @abc.abstractmethod
    def get(self, thing_id: str) -> ThingRecord | None:
        ...


def bind_thing_data_service(binder: Binder):
    """Binds the data service implementation."""
    binder.bind(ThingDataService, to=ThingDataServiceDDB, scope=singleton)


class ThingDataServiceDDB(ThingDataService):
    """DynamoDB implementation of ThingDataService."""

    @inject
    def __init__(self, config: Config, dynamodb):
        self.table = dynamodb.Table(config.example_table_name)

    def create(self, record: ThingRecord) -> None:
        self.table.put_item(Item=record.model_dump())

    def get(self, thing_id: str) -> ThingRecord | None:
        response = self.table.get_item(Key={"thing_id": thing_id})
        if "Item" not in response:
            return None
        return ThingRecord(**response["Item"])

5. Implement the Service

# components/{service_name}/{service_name}/index.py
from injector import Binder, inject, singleton
from service_utils.logger import Logger

from {service_name}_api import (
    CreateThingRequest,
    CreateThingResponse,
    GetThingRequest,
    GetThingResponse,
    ThingService,
)
from {service_name}.config import Config
from {service_name}.data.thing_data_service import ThingDataService, ThingRecord

logger = Logger()


def bind_thing_service(binder: Binder):
    """Binds the service implementation as a singleton."""
    binder.bind(ThingService, to=ThingServiceImpl, scope=singleton)


class ThingServiceImpl(ThingService):
    """Implementation of ThingService."""

    @inject
    def __init__(
        self,
        config: Config,
        data_svc: ThingDataService,
    ):
        self.config = config
        self.data_svc = data_svc

    def create_thing(self, request: CreateThingRequest) -> CreateThingResponse:
        """Creates a new thing with validation."""
        # Generate ID (use your preferred ID generation strategy)
        thing_id = f"thing_{request.name}"

        # Business logic validation
        existing = self.data_svc.get(thing_id)
        if existing:
            raise ValueError(f"Thing with name {request.name} already exists")

        # Create record
        record = ThingRecord(
            thing_id=thing_id,
            name=request.name,
            description=request.description,
        )
        self.data_svc.create(record)

        logger.info(f"Created thing {thing_id}")
        return CreateThingResponse(thing_id=thing_id)

    def get_thing(self, request: GetThingRequest) -> GetThingResponse:
        """Gets a thing by ID."""
        record = self.data_svc.get(request.thing_id)
        if not record:
            raise RecordNotFoundError(f"Thing {request.thing_id} not found")

        return GetThingResponse(
            thing_id=record.thing_id,
            name=record.name,
            description=record.description,
        )

6. Create Test Fixtures

# components/{service_name}/{service_name}/testing/fixtures.py
import pytest
from {service_name}.config import Config


@pytest.fixture
def thing_config():
    """Provides test configuration."""
    return Config(
        environment="test",
        example_table_name="test-things-table",
    )

# components/{service_name}/{service_name}/testing/BUILD
python_test_utils()

7. Create the conftest.py

# components/{service_name}/{service_name}/conftest.py
import os
from unittest.mock import patch, MagicMock

import pytest
from {service_name}.index import ThingServiceImpl

pytest_plugins = [
    "service_utils.testing.fixtures",
    "{service_name}.testing.fixtures",
]


@pytest.fixture(autouse=True)
def mock_env(monkeypatch):
    """Sets up test environment variables."""
    with patch.dict(os.environ, clear=False):
        vars = {
            "EXAMPLE_TABLE_NAME": "test-things-table",
        }
        for k, v in vars.items():
            monkeypatch.setenv(k, v)
        yield os.environ


@pytest.fixture
def mock_data_service():
    """Provides a mock data service for unit tests."""
    return MagicMock()


@pytest.fixture
def thing_service(thing_config, mock_data_service):
    """Provides a service instance for testing."""
    yield ThingServiceImpl(config=thing_config, data_svc=mock_data_service)

8. Write Tests

# components/{service_name}/{service_name}/tests/BUILD
python_tests()

# components/{service_name}/{service_name}/tests/create_thing_test.py
from {service_name}_api import CreateThingRequest


def test_create_thing(thing_service, mock_data_service):
    """Test creating a thing successfully."""
    mock_data_service.get.return_value = None  # No existing record

    request = CreateThingRequest(name="test", description="A test thing")
    response = thing_service.create_thing(request)

    assert response.thing_id == "thing_test"
    mock_data_service.create.assert_called_once()


def test_create_thing_already_exists(thing_service, mock_data_service):
    """Test that creating a duplicate thing raises an error."""
    mock_data_service.get.return_value = ThingRecord(
        thing_id="thing_test",
        name="test",
        description="Existing",
    )

    request = CreateThingRequest(name="test", description="A test thing")
    with pytest.raises(ValueError, match="already exists"):
        thing_service.create_thing(request)

9. Add to Monolith

Add your service bindings to the monolith injector:

# components/monolith/monolith/aws_injector.py
from {service_name}.data.thing_data_service import bind_thing_data_service
from {service_name}.index import bind_thing_service

def create_injector() -> Injector:
    return Injector(
        [
            # ... existing bindings ...
            # Data services
            bind_thing_data_service,
            # Services
            bind_thing_service,
            # ... rest of bindings ...
        ]
    )

10. Create HTTP Router (Optional)

If your service needs HTTP endpoints, create a router in BFF Console:

# components/bff_console/bff_console/routers/things.py
from typing import Annotated

from bff_console.auth import AuthResult
from bff_console.auth.authorizer import Authorizer
from fastapi import APIRouter, Depends
from injector import Binder, inject, singleton

from {service_name}_api import (
    CreateThingRequest,
    GetThingRequest,
    ThingService,
)


class ThingsRouter(APIRouter):
    @inject
    def __init__(self, thing_service: ThingService, authorizer: Authorizer):
        super().__init__()
        self.thing_service = thing_service

        @self.post("/")
        def create_thing(
            body: CreateThingRequest,
            auth: Annotated[AuthResult, Depends(authorizer)],
        ):
            return self.thing_service.create_thing(body)

        @self.get("/{thing_id}")
        def get_thing(
            thing_id: str,
            auth: Annotated[AuthResult, Depends(authorizer)],
        ):
            return self.thing_service.get_thing(GetThingRequest(thing_id=thing_id))


def bind_things_router(binder: Binder):
    binder.bind(ThingsRouter, scope=singleton)

Then add the router to bff_console/app.py:

# In AppModule.provide_app()
routers = {
    # ... existing routers ...
    "/api/things": things_router,
}

And bind the router in the monolith injector.

Key Patterns

Dependency Injection

Use @inject decorator on __init__ methods
Define bind_* functions that bind implementations as singletons
Dependencies are resolved at runtime through the injector hierarchy

Configuration

Inherit from BaseConfig for common fields
Use GCPConfig and AWSConfig for cloud-specific variants
Configuration is loaded from environment variables via Config.load()

Data Layer Abstraction

Define an abstract base class for data access
Create cloud-specific implementations (DDB for AWS, Firestore for GCP)
The implementation is bound in the injector based on the deployment target

Error Handling

Use semantic error types from service_utils.errors
Let exceptions bubble up to FastAPI for HTTP status code mapping
See bff_console/app.py for exception handler examples

Testing

Use pytest fixtures from service_utils.testing.fixtures
Create service-specific fixtures in testing/fixtures.py
Use mocks for unit tests, emulators for integration tests
Register fixtures via pytest_plugins in conftest.py

Checklist

When creating a new microservice:

All of this can be combined in a single large PR, since the structure is well-defined.

Example Services

Reference these existing services for patterns:

billing_service - Simple service with external API integration (Stripe)
indexes_service - Complex service with multiple data models and validation
api_keys_service - Service with secrets management integration
users_accounts_service - Service that orchestrates multiple sub-services for users, accounts, and memberships