DEV Community: Takahiro Ikeuchi

I Built an Open-Source Local Google Calendar API and OAuth Server for Testing AI Agents and MCP Tools

Takahiro Ikeuchi — Mon, 19 Oct 2026 12:45:16 +0000

Lately I have been building more Google Calendar integrations for AI agents and AI app plugins, such as MCP servers that check freeBusy, list events, or create and update events. Testing them against the live Google Calendar API is painful.

So I built Calendarpit.

Calendarpit's admin UI: every Calendar API and OAuth request your app sends, with status codes and masked credentials.

Live tests need a Google Cloud project, OAuth client credentials, and consent screens with dedicated test accounts. State drifts between runs, and a buggy agent could spam attendees or wipe out real events.

I wanted a local emulator that works with the official Google SDK, can be inspected in a browser, and resets cleanly between runs.

Calendarpit came from the needs of the development team behind Chronock, an all-in-one calendar app that brings Google Calendar, Outlook calendars, scheduling links, and automatic event copying into one place. It stays a general-purpose tool, and none of Chronock's data or logic is in it.

What Calendarpit Is and What It Is Not

Calendarpit is an MIT-licensed open-source local server for development and testing. One process serves a Google Calendar API v3 subset, a local OAuth 2.0 server, and an admin UI on http://127.0.0.1:4173. It needs no Google account, Google Cloud project, Google-issued credentials, or external database.

The server never calls Google, and the only external request is the admin UI loading the Roboto Mono font from Google Fonts with a system-font fallback. The current version is 0.9.4, and GET /__calendarpit/health still reports "stage": "preview". It is not a full clone: it covers calendars, events, recurrence, free/busy, incremental sync, and webhooks, and returns an explicit error for anything else.

Try It in 30 Seconds

It ships as a GHCR image, a Homebrew formula, and standalone executables on GitHub Releases for macOS, Linux glibc, and Windows.

# Run with Docker
docker run --rm -p 127.0.0.1:4173:4173 ghcr.io/iktakahiro/calendarpit:latest

# Or install with Homebrew and start the server
brew install iktakahiro/tap/calendarpit
calendarpit serve

The CLI saves state to .calendarpit/calendarpit.sqlite in the working directory and binds to loopback by default. Keep it that way, because the admin UI has no authentication.

The Docker image stores state in /data/calendarpit.sqlite, so the --rm example above is disposable unless you mount a volume with -v calendarpit-data:/data. For disposable E2E runs with the CLI:

calendarpit serve --database :memory: --port 0

--port 0 picks a free port and prints it on startup. Parallel workers each need :memory: or their own database file.

Startup output from calendarpit serve and sample curl responses.

Point Your Existing Google SDK at It

Calendarpit does not intercept traffic. Instead, your app points five endpoints at the local origin: /oauth/authorize, /oauth/token, /userinfo, /oauth/revoke, and the Calendar API root. With @googleapis/calendar and google-auth-library:

const origin = 'http://127.0.0.1:4173';
const auth = new OAuth2Client({
  clientId: 'sample-app',
  clientSecret: 'local-sample-secret',
  redirectUri: 'http://127.0.0.1:4175/callback',
  endpoints: {
    oauth2AuthBaseUrl: `${origin}/oauth/authorize`,
    oauth2TokenUrl: `${origin}/oauth/token`,
  },
});
const api = calendar({ version: 'v3', auth, rootUrl: origin });

This snippet covers authorization, token exchange, and the API root. The docs also show oauth2RevokeUrl, the /userinfo request, and the equivalent setup for Python, Go, Kotlin, PHP, C#, and curl. Apart from endpoint configuration, your tests run through the same SDK code paths as production.

Local OAuth 2.0 with Test Users

Many integration bugs live in OAuth: redirects, PKCE, token refresh, and revocation. Mocking the client leaves that path untested.

Calendarpit implements the OAuth flow locally without passwords. On the consent page, you pick a test user from a dropdown and approve or deny the request.

The local consent page: pick a test user, approve or deny.

It checks exact redirect_uri matches, preserves state, supports S256 PKCE, issues refresh tokens for access_type=offline, and serves /userinfo and /oauth/revoke, without issuing id_tokens or full OIDC. The Connections view shows each grant and lets you revoke it to test reauthorization. You can edit OAuth 2.0 clients under Settings â†’ OAuth clients or in the fixture.

Each connection is an OAuth grant: test user, client, and scopes. Revoke invalidates its tokens.

Fixtures and Deterministic Reset

Every test should start from a known calendar state rather than data from a previous run. Calendarpit seeds its workspace from a JSON fixture via --config ./calendarpit.fixture.json or the built-in default at fixtures/default.json:

{
  "users": [
    { "id": "alex", "name": "Alex Morgan", "email": "alex@example.test" },
    { "id": "sam", "name": "Sam Rivera", "email": "sam@example.test" }
  ],
  "calendars": [
    { "id": "alex-work", "userId": "alex", "summary": "Work calendar", "timeZone": "Asia/Tokyo", "primary": true },
    { "id": "alex-personal", "userId": "alex", "summary": "Personal", "timeZone": "Asia/Tokyo", "primary": false },
    { "id": "sam-work", "userId": "sam", "summary": "Work calendar", "timeZone": "America/New_York", "primary": true }
  ],
  "events": [
    {
      "id": "planning",
      "calendarId": "alex-work",
      "summary": "Weekly planning",
      "start": { "dateTime": "2026-10-19T10:00:00+09:00" },
      "end": { "dateTime": "2026-10-19T11:00:00+09:00" }
    }
  ],
  "clients": [
    {
      "id": "sample-app",
      "secret": "local-sample-secret",
      "redirectUris": ["http://127.0.0.1:4175/callback"]
    }
  ]
}

Between tests, one request resets everything:

curl -X POST http://127.0.0.1:4173/__calendarpit/reset \
  -H "Content-Type: application/json" \
  -d '{}'

Resetting restores the startup fixture with the same IDs and clears tokens, connections, watch channels, and history, so tests authorize again after each reset. Settings â†’ Test data offers the same controls and can load another fixture to replace the current data. Reset still returns to the startup fixture.

Settings â†’ Test data: storage mode, fixture loading, and workspace reset.

Test Webhooks Without a Tunnel

Testing Google Calendar events.watch and calendarList.watch push notifications against the live API requires a public HTTPS callback. In local development, that usually means running a tunnel and waiting on deliveries you cannot control.

With Calendarpit, you register a channel pointing at a local callback using an access token from the OAuth flow:

curl -X POST http://127.0.0.1:4173/calendar/v3/calendars/primary/events/watch \
  -H "Authorization: Bearer <access-token>" \
  -H "Content-Type: application/json" \
  -d '{
    "id": "channel-1",
    "type": "web_hook",
    "address": "http://127.0.0.1:4175/webhooks/calendar",
    "token": "client-verification-token"
  }'

Calendarpit sends a sync notification when the channel is created and an exists notification on each change until the channel expires or channels.stop is called. Only loopback hosts are allowed by default, and CALENDARPIT_WATCH_HOSTS adds other hosts such as a Docker Compose service. Each notification gets up to 3 delivery attempts, and tests can call POST /__calendarpit/deliveries/drain instead of sleep() to wait for pending deliveries.

Monitoring active watch channels and inspecting webhook delivery attempts in the admin UI.

How It Is Built: Architecture, Bun, and Storage

Calendarpit is a Turborepo monorepo built on Bun 1.4.3: runtime, package manager, bun:test, bun check for type checking, HTTP server, and the standalone build (Bun.build with compile). Oxlint and Oxfmt handle linting and formatting.

Workspace boundaries and request routing across the Calendarpit monorepo.

packages/server: the framework-independent Calendar API, OAuth, and watch logic.
apps/calendarpit: the React admin UI, TanStack Start adapter, Bun CLI, and packaging.
apps/sample: a separate Bun app that uses the official @googleapis/calendar and google-auth-library as an HTTP integration test.
apps/docs: the documentation and landing page.

createCalendarpit() exposes an async handler for Web Request objects. It returns a response for API, OAuth, and management routes, or null for UI routes. The TanStack Start entry in apps/calendarpit/src/server.ts tries it first:

fetch: async (request, options) =>
  (await runtime.handle(request)) ?? handleStartRequest(request, options)

Server code is grouped by feature with role suffixes: -routes.ts is the HTTP adapter, -operations.ts changes workspace state as one unit of work, and plain rule modules hold validation and calculations. Dependencies point inward, enforced by architecture.test.ts and Oxlint's cycle check. AGENTS.md files at the root and in each workspace document these rules for contributors and coding agents.

I skipped full DDD or onion layers. An emulator's domain model is the Google Calendar API v3 contract, so entity classes, DTO mappers, repository interfaces, and a DI container would only add indirection.

State lives in memory and is checkpointed as a versioned JSON snapshot into one SQLite row via bun:sqlite, with WAL, FULL sync, and a lock file so two instances cannot overwrite the same database. :memory: writes no files.

Time math uses Temporal and the rrule library. The clock and webhook delivery are injected, so Calendarpit's own expiry and retry tests need no sleeps.

The Admin UI and the TanStack Ecosystem

When an E2E run or an agent misbehaves, I want to see the calendar and the exact HTTP calls side by side. That is what the admin UI is for.

The week schedule view in the admin UI showing events across local test calendars.

It runs on React 19, Vite 8 on Bun, Tailwind CSS 4, Gravity UI Icons, and TanStack Start, Router, Query, and Form:

TanStack Start and Router: SSR and file-based routes for /, /users, /calendars, /connections, /watch, /history, and /settings. The selected user, calendar, date, and view live in search params, so a reload or a shared link restores the same view.
TanStack Query: reads the management API and polls every 2 seconds, so events your app or agent writes through the Calendar API appear without a reload.
TanStack Form: the event, user, calendar, and OAuth client editors, each with one form-level validator.

Creating and editing calendar events in the admin UI using TanStack Form.

The UI has limits too. Dragging events to move or resize them in the calendar grid is not implemented yet, so time changes go through the event editor. The editor covers basic fields, while recurrence, attendees, and conference data are set through the API or fixtures.

Building the Interface with Cloudflare Kumo

For UI components, the admin UI uses @cloudflare/kumo 2.14.0, Cloudflare's MIT-licensed open-source React component library. The community loves shadcn/ui and it is great, but I prefer components I can use as a library rather than copy and paste.

My usual pick is HeroUI, and I kept reaching for it every time. So I looked for a component kit built on a headless UI library and found Kumo, made by Cloudflare and built on Base UI. Kumo (é›²) means "cloud" in Japanese, and as a Japanese developer I felt an affinity for it.

Kumo shows up in 24 source files, from Button and Select to DatePicker, Table, Dialog, Toast, and CommandPalette. Wiring it into Tailwind CSS 4 takes three lines in app.css:

@import '@cloudflare/kumo/styles/tailwind';
@import 'tailwindcss';
@source '../node_modules/@cloudflare/kumo/dist/**/*.{js,jsx,ts,tsx}';

Kumo has no time picker, so the event editor pairs Kumo's DatePicker and Popover with a native <input type="time"> themed by Kumo's Input. Phosphor Icons is installed only as a Kumo peer dependency, while the app's own icons come from Gravity UI Icons.

Kumo DatePicker, Popover, Input, and Select components working together inside the event editor.

Cmd+K / Ctrl+K opens Kumo's CommandPalette to jump between sections, switch test users or calendars, or go to a date.

Navigating views and actions with Kumo's CommandPalette.

On top of Kumo, I added a "timing-bench" theme with warm paper surfaces, near-black ink, monospace readouts, and one signal color for selection and the "now" line. A dark theme follows your system setting by default.

The week schedule view rendered in the dark timing-bench theme.

Testing Strategy, Supply-Chain Precautions, and Releases

If your tests rely on Calendarpit, Calendarpit itself has to be well tested. The repository has 61 bun:test files and 5 Playwright E2E files, organized in layers:

Layer	Runner & Scope
Rules & Operations	`bun:test` with `test.each` on pure validation/recurrence functions, plus in-memory workspace operations using a fixed clock.
HTTP Contract & Lifecycle	Direct `app.handle(Request)` contract assertions and temporary SQLite file tests verifying server restarts, tokens, and watch persistence.
Official SDK (`apps/sample`)	Integration tests using the official `@googleapis/calendar` and `google-auth-library` clients over HTTP.
Browser E2E (Playwright)	Runs against the built CLI (`--database :memory:`), resetting fixtures before each test and submitting OAuth consent in a real browser.
Binary & Docker Smoke	Tests the compiled executable in an isolated directory, plus the container's non-root runtime, HTTP/UI, volume persistence, reset, and memory isolation.

Dependencies are pinned to exact versions. As a supply-chain precaution, bunfig.toml only resolves package versions published at least 48 hours ago, with a matching 2-day cooldown for npm-check-updates.

CI runs bun run validate, builds and smoke-tests the binary, and runs the Playwright suite. The release workflow validates native executables per OS and CPU, publishes checksummed archives to GitHub Releases, smoke-tests and pushes the GHCR image, and updates the Homebrew formula.

Give It a Try

If you are building a Google Calendar integration, an MCP server, or an AI agent that works with calendar events, give Calendarpit a try locally or in CI:

docker run --rm -p 127.0.0.1:4173:4173 ghcr.io/iktakahiro/calendarpit:latest

GitHub Repository: https://github.com/iktakahiro/calendarpit
Documentation (English): https://calendarpit.dev
Documentation (Japanese): https://calendarpit.dev/ja/

If it saves you time, a GitHub star helps others find it. Issues and feedback are welcome, especially on which endpoints, query parameters, or recurrence patterns you need next.

Rethinking Architecture in the AI Era — Part 1: Repository Management

Takahiro Ikeuchi — Mon, 16 Mar 2026 06:24:31 +0000

This is the first installment of a series on software product architecture for the AI age. There's no shortage of content about how fast AI can accelerate prototyping — but I believe medium-term maintainability deserves equal attention. (If this ends up being a one-off article, my apologies in advance.)

Today's topic: repository management. My conclusion upfront: go with a monorepo.

What Is a Monorepo?

For those new to the concept, a monorepo is the practice of managing multiple applications within a single repository. "Multiple apps" can mean any of the following:

A frontend and a backend living together
Full-stack apps for Service A and Service B side by side
Application code alongside shared libraries

A quick aside on microservices: while monorepos are sometimes discussed in that context, I don't have hands-on experience combining the two, so I'll leave that out of scope here. My instinct is that getting the most out of a monorepo requires keeping your tech stack reasonably unified — which arguably conflicts with what microservices are trying to achieve. The monorepo style I'm describing here probably isn't a great fit for microservices.

All that said, rather than dwelling on definitions, the practical answer is simple: use Turborepo.

What Is Turborepo?

Turborepo has become the de facto standard for monorepo setups in TypeScript projects. While it's technically positioned as a general-purpose build tool, in practice you reach for it when adopting a monorepo (or when you want smarter build caching).

Turborepo

Turborepo is a build system optimized for JavaScript and TypeScript, written in Rust.

turborepo.dev

Here's a representative directory structure for a web frontend + backend project using Turborepo:

my-turborepo/
├── apps/
│   ├── web/                  # Frontend app (e.g. Next.js)
│   │   ├── src/
│   │   ├── package.json
│   │   └── tsconfig.json
│   └── api/                  # Backend app (e.g. Hono, Express)
│       ├── src/
│       ├── package.json
│       └── tsconfig.json
├── packages/
│   ├── ui/                   # Shared UI component library
│   │   ├── src/
│   │   ├── package.json
│   │   └── tsconfig.json
│   ├── tsconfig/             # Shared TypeScript config
│   │   ├── base.json
│   │   ├── nextjs.json
│   │   └── package.json
│   └── utils/                # Shared utilities
│       ├── src/
│       ├── package.json
│       └── tsconfig.json
├── .claude/
│   ├── CLAUDE.md
│   └── skills/
├── .codex/
│   └── skills/
├── turbo.json
├── biome.json
├── package.json
├── pnpm-workspace.yaml
└── tsconfig.json

The pattern is straightforward: apps/ holds your individual applications; packages/ holds shared code — utilities, UI components, and config files like tsconfig.

You could technically replicate this directory structure without Turborepo, but Turborepo adds real value by reading each package.json, running builds and tests in bulk or in parallel, and handling build cache intelligently — a significant advantage for larger projects where build times start to hurt.

Why a Monorepo in the AI Era?

Knowledge base management is increasingly recognized as critical in AI-era product development — and that knowledge base naturally includes your codebase. A knowledge base only has value if it's actually referenced. And to be referenced easily, knowledge needs to be close together.

With a monorepo, the moment you clone a single GitHub repository, you have everything in one place: application code, development documentation, agent skills — the full knowledge base for your product. You could devise ways to help AI recognize multiple repositories as belonging to the same project, but why bother? A monorepo gives you that coherence for free.

A common objection goes something like: "Frontend and backend only need to communicate via API contracts — mixing their contexts is unnecessary." I'd argue this mental model is worth updating. In the age of AI coding, the leverage comes from architectures that let you see the full picture and move fast — not from strict context separation.

Conditions for Monorepo Success

Monorepos pair well with AI-era development, but a few conditions make them work significantly better.

1. Your backend is written in TypeScript

A mixed stack — TypeScript frontend, Python backend — can technically be a monorepo, but the shared packages/ become frontend-only, narrowing their value. You also end up managing linting, formatting, and tooling for two different languages.

Worth noting: Turborepo is built around TypeScript (and package.json as the primary unit of configuration), so aligning your stack on TypeScript is the path of least resistance here too.

2. Your review process is adapted for AI-assisted development

In a monorepo with AI coding, implementing a single feature often means the AI writes frontend and backend code in one go. Pull requests naturally reflect that end-to-end implementation. You could split them up — but if you're still enforcing pre-AI-era review rituals, humans become the bottleneck. At minimum, integrating AI-assisted code review is necessary; a broader rethink of your PR workflow is worth considering.

A Real Example: Chronock

My own app, Chronock, is built as a monorepo with TypeScript across the full stack. Here's the structure:

chronock/
├── apps/
│   ├── chronock-app/              # Frontend (React + TanStack Start)
│   ├── chronock-backend/          # Backend (Bun + Connect RPC)
│   ├── chronock-book/             # Booking page (React + TanStack Start)
│   └── chronock-www/              # Marketing site (Astro)
│
├── packages/
│   ├── contract/                  # Proto definitions & generated code
│   ├── i18n/                      # Internationalization utilities
│   └── typescript-config/         # Shared TypeScript config
│
├── specs/                         # Product specifications
│
├── .claude/
│   ├── CLAUDE.md
│   └── skills/
├── .codex/
│   └── skills/
├── biome.jsonc                    # Linter / Formatter config
├── bunfig.toml                    # Bun config
├── docker-compose.local.yml
├── package.json
└── turbo.json                     # Turborepo config

A few things worth highlighting:

The marketing site (Astro) lives in the monorepo alongside the app. This makes tasks like "reflect new feature messaging on the website" or "check for inconsistencies between the latest implementation and the landing page copy" natural things to hand off to an AI agent.
Frontend–backend communication uses Connect RPC (Protocol Buffers). The .proto definition files live in packages/contract, making it seamless to manage the interface definition and run codegen for both frontend and backend from one place.
Product specs and agent skills live in the same repository, which makes coding agents noticeably more effective — they have the context they need without jumping across repos.

Summary

Architecture choices for the AI era:

Centralize context in a monorepo and treat it as a living knowledge base
Aligning your stack on TypeScript pays off in multiple ways
The pull request problem doesn't solve itself — it needs its own dedicated solution

That's a wrap for Part 1.

How to get started DDD & Onion-Architecture in Python web application

Takahiro Ikeuchi — Fri, 22 Apr 2022 11:29:23 +0000

I published the following code for sharing knowledge of DDD & Onion Architecture in Python web applications.

iktakahiro / dddpy

Python DDD & Onion Architecture Example and Techniques

Python DDD & Onion-Architecture Example and Techniques

English | 日本語

Go implementation: oniongo

NOTE: This repository is an example to demonstrate "how to implement DDD architecture in a Python web application." If you use this as a reference, ensure to implement authentication and security before deploying it to a real-world environment!

DeepWiki powered by Devin: https://deepwiki.com/iktakahiro/dddpy
My blog post: https://iktakahiro.dev/python-ddd-onion-architecture

Tech Stack

Prerequisites

Python 3.13 or higher
uv - Python package installer and resolver

Project Setup

Install dependencies using uv:

make install

Run the web app

make dev

Code Architecture

The directory structure is based on Onion Architecture:

├── main.py
├── dddpy
│   ├── domain
│   │   └── todo
│   │       ├── entities
│   │       │   └── todo.py
│   │       ├── value_objects
│   │       │   ├── todo_title.py
│   │       │   ├── todo_description.py
│   │       │   ├── todo_id.py
│   │       │   └── todo_status.py
│

…

View on GitHub

I've also written a git of it on README. But in some ways, DDD is too difficult for us to understand; I would like to explain this architecture.

Motivation

My day job is to develop apps for smartphones using Flutter. In this development, I've adopted DDD with Onion architecture. This approach has worked well so far. On the other hand, the requirements of a native application are more complex than for typical web apps. So, I wanted to see what would happen if I applied DDD to a thin web app.

Conclusion

DDD can be adopted in the Python world as well.
For users familiar with MVC through prominent frameworks such as Django, this codebase may seem bizarre. But I believe it will help in many ways, especially to ensure maintainability and testability.
The variety of interfaces that make up DDD is not necessarily compatible with the dynamically typed language. However, it is achievable using the standard features provided by Python.

Overview

Code Architecture

DDD & Onion Architecture

├── main.py
├── dddpy
│   ├── domain
│   │   └── book
│   │       ├── book.py  # Entity
│   │       ├── book_exception.py  # Exception definitions
│   │       ├── book_repository.py  # Repository interface
│   │       └── isbn.py  # Value Object
│   ├── infrastructure
│   │   └── sqlite
│   │       ├── book
│   │       │   ├── book_dto.py  # DTO using SQLAlchemy
│   │       │   ├── book_query_service.py  # Query service implementation
│   │       │   └── book_repository.py  # Repository implementation
│   │       └── database.py
│   ├── presentation
│   │   └── schema
│   │       └── book
│   │           └── book_error_message.py
│   └── usecase
│       └── book
│           ├── book_command_model.py  # Write models including schemas of the RESTFul API
│           ├── book_command_usecase.py
│           ├── book_query_model.py  # Read models including schemas
│           ├── book_query_service.py  # Query service interface
│           └── book_query_usecase.py
└── tests

Tech Stack

fastapi.tiangolo.com

SQLAlchemy - The Database Toolkit for Python

The Database Toolkit for Python

sqlalchemy.org

Techniques

Repository pattern
Data Transfer Object
CQRS pattern
Unit of Work pattern
Dependency Injection (using FastAPI feature)

How to implement components of DDD

The assumption is that you are familiar with Eric Evans' book and Onion architecture, so I will not explain them. In the following, I will explain how you can implement each DDD component in the Python world.

Entity

To represent an Entity in Python, use __eq__() method to ensure the object's identity.

https://github.com/iktakahiro/dddpy/blob/main/dddpy/domain/book/book.py

class Book:
    def __init__(
        self,
        id: str,
        title: str,
    ):
        self.id: str = id
        self.title: str = title

    def __eq__(self, o: object) -> bool:
        if isinstance(o, Book):
            return self.id == o.id

        return False

In this example, it's anemic. You have to implement more behaviors. I would add that the use of getter and setter is not Pythonic code. On the other hand, I think we can use it if necessary.

Value Object

As I understand it, the requirements for Value Object are the following:

It's not an Entity
If all the properties are the same, The objects are identical.
As a result, they are interchangeable.

To represent a Value Object, use @dataclass decorator with eq=True and frozen=True.

https://github.com/iktakahiro/dddpy/blob/main/dddpy/domain/book/isbn.py

@dataclass(init=False, eq=True, frozen=True)
class Isbn:
    """Isbn represents an ISBN code as a value object"""

    value: str

    def __init__(self, value: str):
        if pattern.match(value) is None:
            raise ValueError("isbn should be a valid format.")

        object.__setattr__(self, "value", value)

Value Objects do not necessarily have to be immutable, but it is better to protect robustness.

Interface (abstract Class)

On a duck-typing language, IMO, it does not go well with defining interfaces. But Abstract Base class helps us to prepare them.

https://github.com/iktakahiro/dddpy/blob/main/dddpy/domain/book/book_repository.py

class BookRepository(ABC):
    @abstractmethod
    def create(self, book: Book) -> Optional[Book]:
        raise NotImplementedError

    @abstractmethod
    def find_by_id(self, id: str) -> Optional[Book]:
        raise NotImplementedError

Program to an interface, not an implementation.

Data Transfer Object and a factory method

DTO (Data Transfer Object) is a good practice to isolate domain objects from the infrastructure layer.

https://github.com/iktakahiro/dddpy/blob/main/dddpy/infrastructure/sqlite/book/book_dto.py

class BookDTO(Base):
    __tablename__ = "book"
    id: Union[str, Column] = Column(String, primary_key=True, autoincrement=False)
    title: Union[str, Column] = Column(String, nullable=False)

On a minimum MVC architecture, models often inherit a base class provided by an O/R Mapper. But in that case, the domain layer would be dependent on the outer layer.

To solve this problem, we can set two rules:

Domain layer classes (such as an Entity or a Value Object) DO NOT extend the SQLAlchemy Base class.
Data transfer Objects extend the O/R mapper class.

Those, including my friend, who assert a DTO and a repository pattern are an overzealous approach.

I'm trying to tell you that you are not using the repository pattern to replace RDBs in the future. You have to take it to ensure testability today.

Read/Write models and request validation

It is helpful to separate a read-only model from a write-only model to implement the CQRS pattern.

I hadn't anticipated this, but this approach also worked well for managing input/output validation.

class BookReadModel(BaseModel):
    id: str = Field(example="vytxeTZskVKR7C7WgdSP3d")
    isbn: str = Field(example="978-0321125217")

FastAPI can set response_model and request_model. Amazingly, by associating the Read model with the Response model and the Write model with the Request model, API documentation can be generated and validated.

https://github.com/iktakahiro/dddpy/blob/main/main.py#L66

@app.post(
    "/books",
    response_model=BookReadModel,
    status_code=status.HTTP_201_CREATED,
    responses={
        status.HTTP_409_CONFLICT: {
            "model": ErrorMessageBookIsbnAlreadyExists,
        },
    },
)
async def create_book(
    data: BookCreateModel,
    book_command_usecase: BookCommandUseCase = Depends(book_command_usecase),
):
    try:
        book = book_command_usecase.create_book(data)
    except BookIsbnAlreadyExistsError as e:
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=e.message,
        )
    except Exception as e:
        logger.error(e)
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )

    return book

Amazingly, by associating read models with the response_model and write models with the request model, the framework can generate API documentation and validate the request body.

CQRS pattern

I will skip a general explanation of CQRS. Instead, I would like to show the specific sequence in the figure below.

By splitting the Read model and Write model, we can flexibly respond to requests' input and output. My dear seniors educated me about DRY for a long time, but I've come to realize that it's not always absolute.

Unit of Work pattern

In this short project, the missing piece is transaction management. For the gap, the Unit of Work pattern fits almost entirely. But first, I would like to say that this is something that I would not strongly recommend to others, including you, because it is redundant, even for me, who suggested the repository pattern so much. However, I can't think of any other way to do transaction management without the assistance of a framework and its middleware.

https://github.com/iktakahiro/dddpy/blob/main/dddpy/usecase/book/book_command_usecase.py#L18

class BookCommandUseCaseUnitOfWork(ABC):
    book_repository: BookRepository

    @abstractmethod
    def begin(self):
        raise NotImplementedError

    @abstractmethod
    def commit(self):
        raise NotImplementedError

    @abstractmethod
    def rollback(self):
        raise NotImplementedError

https://github.com/iktakahiro/dddpy/blob/main/dddpy/infrastructure/sqlite/book/book_repository.py#L63

class BookCommandUseCaseUnitOfWorkImpl(BookCommandUseCaseUnitOfWork):
    def __init__(
        self,
        session: Session,
        book_repository: BookRepository,
    ):
        self.session: Session = session
        self.book_repository: BookRepository = book_repository

    def begin(self):
        self.session.begin()

    def commit(self):
        self.session.commit()

    def rollback(self):
        self.session.rollback()

Dependency Injection

This is a last topic, Dependency Injection sounds like an exaggeration, but it essentially means assigning a class instance to a property of other classes. One of the reasons I like FastAPI is that it provides a DI mechanism by default.

https://github.com/iktakahiro/dddpy/blob/main/main.py#L53

def book_query_usecase(session: Session = Depends(get_session)) -> BookQueryUseCase:
    book_query_service: BookQueryService = BookQueryServiceImpl(session)
    return BookQueryUseCaseImpl(book_query_service)


def book_command_usecase(session: Session = Depends(get_session)) -> BookCommandUseCase:
    book_repository: BookRepository = BookRepositoryImpl(session)
    uow: BookCommandUseCaseUnitOfWork = BookCommandUseCaseUnitOfWorkImpl(
        session, book_repository=book_repository
    )
    return BookCommandUseCaseImpl(uow)

I don't think there is a widely known methodology for practicing DDD in Python. If you are going to try it, please refer to it.