Chapter 30 — Capstone: Rebuilding a TypeScript Service in Professional Python
1. Capstone goal
You will rebuild a simplified TypeScript order-processing service in Python.
The objective is not a line-by-line translation.
The objective is to preserve business behavior while redesigning the implementation around Python's semantics and ecosystem.
You will practice:
- domain modeling;
- typing;
- protocols;
- dataclasses;
- context managers;
- exceptions;
- repositories;
- unit of work;
- HTTP boundaries;
- async workers;
- outbox publishing;
- idempotency;
- testing;
- packaging;
- observability;
- deployment;
- English technical communication.
2. Original TypeScript service
Assume the original service uses:
TypeScript
Node.js
NestJS-style dependency injection
PostgreSQL
ORM entities
REST API
Message broker
Payment provider
Jest
Docker
The existing architecture:
Controller
↓
OrderService
↓
ORM Repository
↓
PostgreSQL
OrderService
↓
Payment SDK
OrderService
↓
Broker SDK
Known problems:
- ORM entities contain API decorators;
- payment occurs inside the main service method;
- message publication happens after database commit without an outbox;
- retries can duplicate charges;
- tests mock ORM internals;
- service class contains all rules;
- domain state can be changed directly;
- request DTOs leak into persistence;
- module initialization creates clients;
- background publishing has no graceful shutdown.
3. Migration principles
Principle 1: Preserve behavior before changing design
Create contract tests around the existing service.
Principle 2: Migrate by vertical slice
Move one use case at a time.
Principle 3: Do not share database tables casually
Choose a clear transition strategy.
Principle 4: Keep external contracts stable initially
Avoid simultaneous API redesign and language migration.
Principle 5: Observe both systems
Compare logs, metrics, outputs, and side effects.
Principle 6: Define rollback
Every migration step needs a safe reversal.
Principle 7: Do not translate framework patterns mechanically
Re-evaluate every abstraction.
4. Capstone feature set
The service supports:
- create customer;
- create draft order;
- add order line;
- place order;
- authorize payment;
- mark order paid;
- publish order events;
- retrieve order;
- cancel unpaid order;
- retry failed outbox messages.
Non-functional requirements:
- idempotent payment authorization;
- explicit timeouts;
- transactional outbox;
- structured logs;
- metrics;
- health checks;
- graceful shutdown;
- tests at several layers;
- package and deploy as an immutable artifact.
5. Project structure
python-order-service/
├── pyproject.toml
├── README.md
├── src/
│ └── order_service/
│ ├── __init__.py
│ ├── __main__.py
│ ├── bootstrap.py
│ ├── settings.py
│ ├── domain/
│ │ ├── customer.py
│ │ ├── order.py
│ │ ├── money.py
│ │ ├── events.py
│ │ └── errors.py
│ ├── application/
│ │ ├── commands.py
│ │ ├── queries.py
│ │ ├── handlers.py
│ │ ├── ports.py
│ │ └── results.py
│ ├── infrastructure/
│ │ ├── postgres/
│ │ │ ├── repositories.py
│ │ │ ├── unit_of_work.py
│ │ │ └── mappings.py
│ │ ├── payments/
│ │ │ └── client.py
│ │ ├── messaging/
│ │ │ ├── outbox.py
│ │ │ └── publisher.py
│ │ └── observability/
│ └── delivery/
│ ├── http/
│ │ ├── routes.py
│ │ ├── schemas.py
│ │ └── errors.py
│ └── workers/
│ └── outbox_worker.py
└── tests/
├── unit/
├── component/
├── integration/
├── contract/
└── acceptance/
6. Milestone 1 — Value objects
Order ID
@dataclass(frozen=True)
class OrderId:
value: str
def __post_init__(
self,
) -> None:
normalized = (
self.value.strip()
)
if not normalized:
raise ValueError(
"order ID must not be blank"
)
object.__setattr__(
self,
"value",
normalized,
)
Customer ID
@dataclass(frozen=True)
class CustomerId:
value: str
Product ID
@dataclass(frozen=True)
class ProductId:
value: str
Quantity
@dataclass(frozen=True)
class Quantity:
value: int
def __post_init__(
self,
) -> None:
if self.value <= 0:
raise ValueError(
"quantity must be positive"
)
Money
@dataclass(frozen=True)
class Money:
amount_cents: int
currency: str
def __post_init__(
self,
) -> None:
normalized = (
self.currency
.strip()
.upper()
)
if len(normalized) != 3:
raise ValueError(
"invalid currency"
)
object.__setattr__(
self,
"currency",
normalized,
)
def multiply(
self,
quantity: Quantity,
) -> "Money":
return Money(
self.amount_cents
* quantity.value,
self.currency,
)
7. Milestone 2 — Order line
@dataclass(frozen=True)
class OrderLine:
product_id: ProductId
quantity: Quantity
unit_price: Money
@property
def total(self) -> Money:
return (
self.unit_price
.multiply(
self.quantity
)
)
A line is immutable.
To change quantity, create a new line through an aggregate operation.
8. Milestone 3 — Domain events
@dataclass(frozen=True)
class OrderPlaced:
order_id: OrderId
customer_id: CustomerId
total: Money
occurred_at: datetime
@dataclass(frozen=True)
class OrderPaid:
order_id: OrderId
authorization_id: str
occurred_at: datetime
Events are immutable completed facts.
9. Milestone 4 — Order aggregate
class Order:
def __init__(
self,
*,
order_id: OrderId,
customer_id: CustomerId,
status: OrderStatus = (
OrderStatus.DRAFT
),
lines: tuple[
OrderLine,
...
] = (),
) -> None:
self._order_id = order_id
self._customer_id = (
customer_id
)
self._status = status
self._lines = list(lines)
self._events: list[
object
] = []
self._authorization_id: (
str | None
) = None
Properties:
@property
def order_id(
self,
) -> OrderId:
return self._order_id
@property
def status(
self,
) -> OrderStatus:
return self._status
@property
def lines(
self,
) -> tuple[
OrderLine,
...
]:
return tuple(
self._lines
)
10. Adding a line
def add_line(
self,
*,
product_id: ProductId,
quantity: Quantity,
unit_price: Money,
) -> None:
self._require_draft()
existing_index = next(
(
index
for index, line
in enumerate(
self._lines
)
if (
line.product_id
== product_id
)
),
None,
)
new_line = OrderLine(
product_id=product_id,
quantity=quantity,
unit_price=unit_price,
)
if existing_index is None:
self._lines.append(
new_line
)
else:
self._lines[
existing_index
] = new_line
The aggregate controls replacement.
11. Calculating total
@property
def total(self) -> Money:
if not self._lines:
return Money(
0,
"USD",
)
currency = (
self._lines[
0
].unit_price.currency
)
result = Money(
0,
currency,
)
for line in self._lines:
result = (
result + line.total
)
return result
A production model should define currency policy explicitly.
For this capstone, all lines in one order must share one currency.
12. Placing the order
def place(
self,
*,
occurred_at: datetime,
) -> None:
self._require_draft()
if not self._lines:
raise EmptyOrderError(
self._order_id
)
self._status = (
OrderStatus.PLACED
)
self._events.append(
OrderPlaced(
order_id=self._order_id,
customer_id=(
self._customer_id
),
total=self.total,
occurred_at=occurred_at,
)
)
13. Marking paid
def mark_paid(
self,
*,
authorization_id: str,
occurred_at: datetime,
) -> None:
if (
self._status
is OrderStatus.PAID
):
if (
self._authorization_id
== authorization_id
):
return
raise OrderAlreadyPaidError(
self._order_id
)
if (
self._status
is not OrderStatus.PLACED
):
raise InvalidOrderStateError(
self._order_id,
self._status,
)
self._authorization_id = (
authorization_id
)
self._status = (
OrderStatus.PAID
)
self._events.append(
OrderPaid(
order_id=self._order_id,
authorization_id=(
authorization_id
),
occurred_at=occurred_at,
)
)
The idempotent same-authorization case returns safely.
14. Pulling domain events
def pull_events(
self,
) -> tuple[object, ...]:
events = tuple(
self._events
)
self._events.clear()
return events
Only application infrastructure should call this after domain operations.
15. Milestone 5 — Commands
@dataclass(frozen=True)
class CreateOrder:
order_id: OrderId
customer_id: CustomerId
@dataclass(frozen=True)
class AddOrderLine:
order_id: OrderId
product_id: ProductId
quantity: Quantity
@dataclass(frozen=True)
class PlaceOrder:
order_id: OrderId
@dataclass(frozen=True)
class AuthorizeOrderPayment:
order_id: OrderId
idempotency_key: str
Commands represent application intentions.
16. Milestone 6 — Ports
class OrderRepository(
Protocol
):
def get(
self,
order_id: OrderId,
) -> Order | None:
...
def add(
self,
order: Order,
) -> None:
...
class ProductCatalog(
Protocol
):
def get_price(
self,
product_id: ProductId,
) -> Money:
...
class Clock(
Protocol
):
def now(
self,
) -> datetime:
...
class PaymentGateway(
Protocol
):
def authorize(
self,
*,
idempotency_key: str,
order_id: OrderId,
amount: Money,
) -> PaymentAuthorization:
...
17. Payment authorization result
@dataclass(frozen=True)
class PaymentAuthorization:
authorization_id: str
approved: bool
decline_reason: (
str | None
) = None
A declined payment is an expected result, not necessarily an infrastructure exception.
Network failure remains an exception.
18. Unit of work
class UnitOfWork(
Protocol
):
orders: OrderRepository
outbox: OutboxRepository
idempotency: (
IdempotencyRepository
)
def __enter__(
self,
) -> "UnitOfWork":
...
def __exit__(
self,
exception_type,
exception,
traceback,
) -> bool:
...
def commit(self) -> None:
...
Use one instance per application operation.
19. Create order handler
class CreateOrderHandler:
def __init__(
self,
*,
unit_of_work_factory,
) -> None:
self._unit_of_work_factory = (
unit_of_work_factory
)
def handle(
self,
command: CreateOrder,
) -> None:
with (
self._unit_of_work_factory()
as unit_of_work
):
existing = (
unit_of_work.orders
.get(command.order_id)
)
if existing is not None:
raise DuplicateOrderError(
command.order_id
)
order = Order(
order_id=command.order_id,
customer_id=(
command.customer_id
),
)
unit_of_work.orders.add(
order
)
unit_of_work.commit()
20. Add line handler
class AddOrderLineHandler:
def __init__(
self,
*,
unit_of_work_factory,
product_catalog: (
ProductCatalog
),
) -> None:
self._unit_of_work_factory = (
unit_of_work_factory
)
self._product_catalog = (
product_catalog
)
def handle(
self,
command: AddOrderLine,
) -> None:
price = (
self._product_catalog
.get_price(
command.product_id
)
)
with (
self._unit_of_work_factory()
as unit_of_work
):
order = (
unit_of_work.orders
.get(command.order_id)
)
if order is None:
raise OrderNotFoundError(
command.order_id
)
order.add_line(
product_id=(
command.product_id
),
quantity=(
command.quantity
),
unit_price=price,
)
unit_of_work.commit()
The catalog lookup occurs outside the database transaction unless consistency requirements demand otherwise.
21. Place order handler
class PlaceOrderHandler:
def __init__(
self,
*,
unit_of_work_factory,
clock: Clock,
) -> None:
self._unit_of_work_factory = (
unit_of_work_factory
)
self._clock = clock
def handle(
self,
command: PlaceOrder,
) -> None:
with (
self._unit_of_work_factory()
as unit_of_work
):
order = (
unit_of_work.orders
.get(command.order_id)
)
if order is None:
raise OrderNotFoundError(
command.order_id
)
order.place(
occurred_at=(
self._clock.now()
)
)
add_events_to_outbox(
order.pull_events(),
unit_of_work.outbox,
)
unit_of_work.commit()
22. Payment workflow challenge
A remote payment call cannot share the database transaction.
A dangerous design:
with unit_of_work:
payment_gateway.authorize(...)
order.mark_paid(...)
unit_of_work.commit()
If payment succeeds but commit fails, the order remains unpaid locally.
The capstone uses an idempotency record and reconciliation strategy.
23. Payment handler — phase 1
Inside a short transaction:
- load order;
- verify payable state;
- inspect idempotency record;
- create pending payment attempt;
- commit.
Then call payment gateway outside the transaction.
Then open a second transaction to record the result.
This reduces long transactions and supports recovery.
24. Payment attempt model
@dataclass(frozen=True)
class PaymentAttempt:
idempotency_key: str
order_id: OrderId
amount: Money
status: PaymentAttemptStatus
authorization_id: (
str | None
) = None
Statuses:
class PaymentAttemptStatus(
Enum
):
PENDING = "pending"
APPROVED = "approved"
DECLINED = "declined"
A production design also needs timestamps and failure metadata.
25. Authorize payment handler
class AuthorizePaymentHandler:
def __init__(
self,
*,
unit_of_work_factory,
payment_gateway: (
PaymentGateway
),
clock: Clock,
) -> None:
self._unit_of_work_factory = (
unit_of_work_factory
)
self._payment_gateway = (
payment_gateway
)
self._clock = clock
Prepare:
def _prepare(
self,
command: (
AuthorizeOrderPayment
),
) -> PaymentAttempt:
with (
self._unit_of_work_factory()
as unit_of_work
):
existing = (
unit_of_work.idempotency
.get(
command.idempotency_key
)
)
if existing is not None:
return existing
order = (
unit_of_work.orders
.get(command.order_id)
)
if order is None:
raise OrderNotFoundError(
command.order_id
)
attempt = PaymentAttempt(
idempotency_key=(
command.idempotency_key
),
order_id=order.order_id,
amount=order.total,
status=(
PaymentAttemptStatus
.PENDING
),
)
unit_of_work.idempotency.add(
attempt
)
unit_of_work.commit()
return attempt
26. Remote authorization
def _authorize_remote(
self,
attempt: PaymentAttempt,
) -> PaymentAuthorization:
return (
self._payment_gateway
.authorize(
idempotency_key=(
attempt.idempotency_key
),
order_id=(
attempt.order_id
),
amount=attempt.amount,
)
)
The same key must be sent to a provider that supports idempotency, or the adapter must provide another safe strategy.
27. Finalizing approval
def _record_approval(
self,
*,
attempt: PaymentAttempt,
authorization: (
PaymentAuthorization
),
) -> None:
with (
self._unit_of_work_factory()
as unit_of_work
):
current_attempt = (
unit_of_work.idempotency
.get(
attempt.idempotency_key
)
)
if (
current_attempt is not None
and current_attempt.status
is (
PaymentAttemptStatus
.APPROVED
)
):
return
order = (
unit_of_work.orders
.get(attempt.order_id)
)
if order is None:
raise OrderNotFoundError(
attempt.order_id
)
order.mark_paid(
authorization_id=(
authorization
.authorization_id
),
occurred_at=(
self._clock.now()
),
)
unit_of_work.idempotency.mark_approved(
attempt.idempotency_key,
authorization.authorization_id,
)
add_events_to_outbox(
order.pull_events(),
unit_of_work.outbox,
)
unit_of_work.commit()
28. Full payment handler
def handle(
self,
command: AuthorizeOrderPayment,
) -> PaymentAuthorization:
attempt = self._prepare(
command
)
if (
attempt.status
is PaymentAttemptStatus.APPROVED
):
return PaymentAuthorization(
authorization_id=(
attempt.authorization_id
or ""
),
approved=True,
)
authorization = (
self._authorize_remote(
attempt
)
)
if authorization.approved:
self._record_approval(
attempt=attempt,
authorization=(
authorization
),
)
else:
self._record_decline(
attempt=attempt,
authorization=(
authorization
),
)
return authorization
A production design must address a crash after remote approval but before local finalization.
Reconciliation uses the persisted pending attempt and the provider idempotency key.
29. Reconciliation job
class ReconcilePaymentsJob:
def run(self) -> None:
pending = (
repository
.find_pending_older_than(
threshold
)
)
for attempt in pending:
provider_result = (
payment_gateway
.lookup(
attempt
.idempotency_key
)
)
if provider_result is None:
continue
finalize(
attempt,
provider_result,
)
This closes the failure gap.
Distributed reliability requires recovery paths, not only happy-path transactions.
30. Outbox message
@dataclass(frozen=True)
class OutboxMessage:
message_id: str
message_type: str
schema_version: int
payload: bytes
occurred_at: datetime
Serialize domain event explicitly:
def serialize_event(
event: object,
) -> OutboxMessage:
match event:
case OrderPlaced():
...
case OrderPaid():
...
case _:
raise TypeError(
f"Unsupported event: "
f"{type(event).__name__}"
)
Use an exhaustive branch and stable schema names.
31. Outbox worker
class OutboxWorker:
def __init__(
self,
*,
repository,
publisher,
poll_interval_seconds: float,
) -> None:
self._repository = repository
self._publisher = publisher
self._poll_interval = (
poll_interval_seconds
)
Async loop:
async def run(
self,
stop_event: asyncio.Event,
) -> None:
while not stop_event.is_set():
messages = await asyncio.to_thread(
self._repository.claim_batch,
100,
)
if not messages:
try:
await asyncio.wait_for(
stop_event.wait(),
timeout=(
self._poll_interval
),
)
except TimeoutError:
pass
continue
for message in messages:
await self._publish_one(
message
)
A fully async database adapter would avoid to_thread.
32. Publishing one message
async def _publish_one(
self,
message: OutboxMessage,
) -> None:
try:
await self._publisher.publish(
message
)
except (
TimeoutError,
ConnectionError,
) as error:
await asyncio.to_thread(
self._repository
.record_failure,
message.message_id,
str(error),
)
return
await asyncio.to_thread(
self._repository.mark_sent,
message.message_id,
)
A duplicate publish remains possible if the process fails after publish but before mark_sent.
Consumers must deduplicate.
33. HTTP schemas
class AddLinePayload(
TypedDict
):
product_id: str
quantity: int
Parser:
def parse_add_line(
order_id: str,
payload: object,
) -> AddOrderLine:
if not isinstance(
payload,
dict,
):
raise ValidationError(
"body must be an object"
)
product_id = payload.get(
"product_id"
)
quantity = payload.get(
"quantity"
)
if not isinstance(
product_id,
str,
):
raise ValidationError(
"product_id must be "
"a string"
)
if (
isinstance(
quantity,
bool,
)
or not isinstance(
quantity,
int,
)
):
raise ValidationError(
"quantity must be "
"an integer"
)
return AddOrderLine(
order_id=OrderId(
order_id
),
product_id=ProductId(
product_id
),
quantity=Quantity(
quantity
),
)
34. HTTP route shape
Framework-neutral pseudocode:
async def add_line_route(
request,
order_id: str,
handler: AddOrderLineHandler,
):
try:
payload = await request.json()
command = parse_add_line(
order_id,
payload,
)
await asyncio.to_thread(
handler.handle,
command,
)
except ValidationError as error:
return response(
status=400,
body={
"code": (
"validation_error"
),
"message": str(error),
},
)
except OrderNotFoundError:
return response(
status=404,
body={
"code": (
"order_not_found"
),
},
)
return response(
status=204,
body=None,
)
A synchronous framework can call the handler directly.
Choose sync or async based on adapter ecosystem and workload.
35. Response model
class OrderResponse(
TypedDict
):
order_id: str
customer_id: str
status: str
currency: str
total_cents: int
lines: list[
"OrderLineResponse"
]
Serializer:
def serialize_order(
order: Order,
) -> OrderResponse:
return {
"order_id": (
order.order_id.value
),
"customer_id": (
order.customer_id.value
),
"status": (
order.status.value
),
"currency": (
order.total.currency
),
"total_cents": (
order.total.amount_cents
),
"lines": [
{
"product_id": (
line.product_id
.value
),
"quantity": (
line.quantity
.value
),
"unit_price_cents": (
line.unit_price
.amount_cents
),
}
for line in order.lines
],
}
Internal fields remain hidden.
36. Persistence mapping
Database row:
@dataclass(frozen=True)
class OrderRow:
order_id: str
customer_id: str
status: str
authorization_id: (
str | None
)
Mapping:
def row_to_order(
row: OrderRow,
line_rows: list[
OrderLineRow
],
) -> Order:
lines = tuple(
OrderLine(
product_id=ProductId(
line.product_id
),
quantity=Quantity(
line.quantity
),
unit_price=Money(
line.unit_price_cents,
line.currency,
),
)
for line in line_rows
)
return Order.restore(
order_id=OrderId(
row.order_id
),
customer_id=CustomerId(
row.customer_id
),
status=OrderStatus(
row.status
),
authorization_id=(
row.authorization_id
),
lines=lines,
)
Use a dedicated restore constructor that does not emit new domain events.
37. Restore constructor
@classmethod
def restore(
cls,
*,
order_id: OrderId,
customer_id: CustomerId,
status: OrderStatus,
authorization_id: (
str | None
),
lines: tuple[
OrderLine,
...
],
) -> "Order":
order = cls(
order_id=order_id,
customer_id=customer_id,
status=status,
lines=lines,
)
order._authorization_id = (
authorization_id
)
return order
Restoration should still preserve invariants.
Do not allow arbitrary invalid database state silently.
38. Optimistic concurrency
Add version:
self._version = version
Update:
UPDATE orders
SET
status = :status,
version = version + 1
WHERE
order_id = :order_id
AND version = :expected_version
If zero rows update, raise:
class ConcurrentOrderUpdateError(
InfrastructureError
):
pass
The application may retry the whole use case when safe.
Do not retry after irreversible external side effects without idempotency.
39. Database migrations
Initial schema:
customers
orders
order_lines
payment_attempts
outbox_messages
consumer_inbox
Migration rules:
- primary keys explicit;
- unique idempotency key;
- foreign keys;
- order version;
- indexes for pending outbox query;
- timestamps stored consistently;
- payload schema version;
- status constraints where practical.
Test migrations from an older production-like schema.
40. Logging
Use event names:
logger.info(
"payment_authorized",
extra={
"order_id": (
order_id.value
),
"authorization_id": (
authorization_id
),
"duration_ms": (
duration_ms
),
},
)
Avoid logging:
- full payment request;
- tokens;
- complete customer records;
- raw provider errors containing sensitive data.
41. Metrics
Suggested metrics:
http_requests_total
http_request_duration_seconds
orders_created_total
orders_placed_total
payments_authorized_total
payments_declined_total
payment_gateway_duration_seconds
outbox_pending_messages
outbox_publish_failures_total
database_transaction_duration_seconds
Use bounded labels:
route
method
status_class
message_type
outcome
Avoid labels containing order ID or customer ID.
42. Tracing
Trace path:
POST /orders/{id}/payment
├── load order
├── create payment attempt
├── call payment provider
├── finalize order
└── write outbox
External calls and database operations deserve spans.
Pure value-object methods generally do not.
43. Settings
@dataclass(frozen=True)
class Settings:
database_url: str
payment_url: str
payment_token: str
payment_timeout_seconds: float
outbox_poll_seconds: float
environment: str
Parser validates:
- required values;
- positive timeouts;
- supported environment;
- safe defaults.
Redact token representation.
44. Bootstrap
def create_runtime(
settings: Settings,
) -> Runtime:
database = create_database(
settings.database_url
)
payment_client = (
create_payment_client(
url=settings.payment_url,
token=(
settings.payment_token
),
timeout=(
settings
.payment_timeout_seconds
),
)
)
unit_of_work_factory = (
create_uow_factory(
database
)
)
payment_gateway = (
RemotePaymentGateway(
payment_client
)
)
handlers = Handlers(
create_order=(
CreateOrderHandler(
unit_of_work_factory=(
unit_of_work_factory
)
)
),
authorize_payment=(
AuthorizePaymentHandler(
unit_of_work_factory=(
unit_of_work_factory
),
payment_gateway=(
payment_gateway
),
clock=SystemClock(),
)
),
)
return Runtime(
handlers=handlers,
database=database,
payment_client=(
payment_client
),
)
One composition root makes dependencies visible.
45. Graceful runtime
async def main() -> int:
settings = load_settings(
os.environ
)
runtime = create_runtime(
settings
)
stop_event = asyncio.Event()
install_signal_handlers(
stop_event
)
async with asyncio.TaskGroup() as group:
group.create_task(
run_http_server(
runtime,
stop_event,
)
)
group.create_task(
runtime.outbox_worker.run(
stop_event
)
)
await runtime.close()
return 0
Real server libraries define their own startup and shutdown APIs.
The principle is scoped ownership.
46. Unit tests
Money:
def test_money_adds_same_currency():
result = (
Money(500, "USD")
+ Money(250, "USD")
)
assert result == Money(
750,
"USD",
)
Order state:
def test_empty_order_cannot_be_placed(
fixed_time,
):
order = create_order()
with pytest.raises(
EmptyOrderError
):
order.place(
occurred_at=fixed_time
)
Idempotent paid state:
def test_mark_paid_is_idempotent_for_same_authorization(
placed_order,
fixed_time,
):
placed_order.mark_paid(
authorization_id="auth-1",
occurred_at=fixed_time,
)
placed_order.mark_paid(
authorization_id="auth-1",
occurred_at=fixed_time,
)
assert (
placed_order.status
is OrderStatus.PAID
)
47. Component tests
Use fake unit of work and payment gateway.
class FakePaymentGateway:
def __init__(
self,
result: PaymentAuthorization,
) -> None:
self._result = result
self.calls: list[
tuple[str, OrderId, Money]
] = []
def authorize(
self,
*,
idempotency_key: str,
order_id: OrderId,
amount: Money,
) -> PaymentAuthorization:
self.calls.append(
(
idempotency_key,
order_id,
amount,
)
)
return self._result
Assert:
- gateway called once;
- order marked paid;
- outbox event written;
- idempotency record finalized.
48. Contract tests
Payment adapter contract:
- request method;
- path;
- authentication;
- idempotency header;
- amount serialization;
- timeout;
- decline parsing;
- transient error mapping;
- malformed response handling.
Repository contract:
- add and get;
- line ordering;
- optimistic version;
- transaction rollback;
- concurrent update;
- nullable authorization;
- event-free restoration.
49. Integration tests
Real database:
- apply migrations;
- create order;
- add lines;
- commit;
- reload aggregate;
- verify mapping;
- test conflict;
- test outbox transaction;
- roll back failure;
- verify indexes for worker query where needed.
Use the production database technology.
50. Acceptance tests
Example scenario:
POST /orders
POST /orders/{id}/lines
POST /orders/{id}/place
POST /orders/{id}/payment
GET /orders/{id}
Assertions:
- status codes;
- response schema;
- state transitions;
- provider call;
- outbox row;
- duplicate payment request returns same result;
- invalid order transition returns stable error.
51. Migration from TypeScript
Phase 1 — Baseline
- document current APIs;
- capture message schemas;
- capture database schema;
- create contract tests;
- add correlation IDs;
- record production metrics.
Phase 2 — Shadow reads
Python service reads the same data and compares responses without serving users.
Phase 3 — One read endpoint
Route a small percentage of read traffic.
Phase 4 — One write use case
Choose a low-risk, idempotent operation.
Phase 5 — Payment workflow
Migrate only after idempotency and reconciliation are proven.
Phase 6 — Event publishing
Use one owner for each event during transition.
Phase 7 — Retire TypeScript path
After stable metrics, remove dual behavior.
52. Database migration strategies
Shared schema
Both services access the same tables.
Risk:
- hidden coupling;
- ORM differences;
- migration coordination;
- write conflicts.
New schema with replication
Cleaner ownership but more migration work.
Strangler facade
One public API routes use cases to old or new implementation.
Often the safest incremental strategy.
Choose according to operational constraints.
53. Dual-write warning
Writing to both systems from one request can create divergence.
If one write succeeds and one fails:
- which is authoritative?
- how is recovery performed?
- how are duplicates handled?
- what is the rollback?
Prefer one source of truth plus events or controlled replication.
54. Shadow comparison
For read operations:
- serve TypeScript response;
- call Python implementation asynchronously;
- normalize nondeterministic fields;
- compare;
- record differences;
- do not expose shadow failure to the user initially.
Protect personal data in comparison logs.
Do not double-call expensive or side-effecting endpoints.
55. Rollout
Use:
- feature flags;
- tenant allowlist;
- percentage routing;
- environment rollout;
- operation-specific routing;
- instant fallback;
- dashboards;
- error budgets.
Observe:
- latency;
- error rate;
- payment success;
- duplicate rate;
- outbox age;
- reconciliation count;
- database load;
- memory;
- CPU.
56. Rollback
A rollback is possible only if data compatibility remains.
Before rollout, define:
- schema compatibility window;
- message compatibility;
- ownership of pending jobs;
- idempotency records;
- routing switch;
- deployment artifact;
- database downgrade policy;
- recovery procedure.
Never discover rollback behavior during the incident.
57. TypeScript-to-Python mapping
Interface
TypeScript:
interface OrderRepository {
get(id: OrderId): Promise<Order | null>;
}
Python:
class OrderRepository(
Protocol
):
def get(
self,
order_id: OrderId,
) -> Order | None:
...
Readonly DTO
Use frozen dataclass or typed dictionary depending on runtime shape.
Promise
Use coroutine only when the adapter is genuinely asynchronous.
Decorator-driven injection
Prefer explicit constructors and composition root.
Exceptions
Create typed exception hierarchy and boundary mapping.
Jest mocks
Prefer fakes, specs, contracts, and dependency injection.
58. Architecture decision records for capstone
Write at least these ADRs:
- framework-neutral domain;
- separate domain and database models;
- transactional outbox;
- idempotent payment attempts;
- explicit composition root;
srclayout and wheel deployment;- sync or async database choice;
- messaging delivery semantics;
- supported Python version;
- observability standard.
59. Production readiness checklist
Correctness
- domain invariants tested;
- state transitions explicit;
- idempotency tested;
- migrations tested;
- serialization versioned;
- failure recovery tested.
Reliability
- timeouts;
- bounded retries;
- outbox;
- reconciliation;
- graceful shutdown;
- readiness;
- queue limits.
Security
- validation;
- authorization;
- secrets;
- dependency review;
- least privilege;
- redacted logs.
Operations
- structured logs;
- metrics;
- traces;
- alerts;
- runbooks;
- dashboards;
- rollback.
Delivery
- wheel or container built;
- clean installation tested;
- configuration documented;
- immutable artifact;
- deployment automation.
60. English presentation task
Prepare a 15-minute presentation:
Rebuilding an order-processing service from TypeScript to Python without losing reliability.
Suggested structure:
- existing problems;
- migration goals;
- Python architecture;
- domain model;
- ports and adapters;
- payment idempotency;
- outbox;
- testing;
- rollout;
- lessons learned.
Record yourself and revise unclear explanations.
61. English writing task
Write a 1,200-word technical design document for the capstone.
Include:
- context;
- goals;
- non-goals;
- architecture;
- domain model;
- persistence;
- external dependencies;
- failure handling;
- observability;
- security;
- testing;
- migration;
- rollout;
- rollback;
- open questions.
62. Final implementation milestones
Milestone A — Domain
- value objects;
- order aggregate;
- events;
- exceptions;
- unit tests.
Milestone B — Application
- commands;
- handlers;
- protocols;
- fake adapters;
- component tests.
Milestone C — Persistence
- schema;
- mappings;
- repositories;
- unit of work;
- migrations;
- integration tests.
Milestone D — Delivery
- HTTP parsers;
- routes;
- error mapping;
- acceptance tests.
Milestone E — Reliability
- idempotency;
- outbox;
- worker;
- reconciliation;
- graceful shutdown.
Milestone F — Operations
- settings;
- logging;
- metrics;
- tracing;
- health checks.
Milestone G — Packaging
pyproject.toml;- wheel;
- clean install;
- deployment artifact.
Milestone H — Migration
- contract tests;
- shadow traffic;
- gradual rollout;
- rollback exercise.
63. Final capstone checkpoint
You have completed the capstone when you can demonstrate:
- a framework-independent domain;
- validated value objects;
- protected aggregate state;
- application handlers;
- protocol-based ports;
- fake and real adapters;
- transaction boundaries;
- idempotent payment handling;
- transactional outbox;
- duplicate-safe consumer behavior;
- HTTP boundary parsing;
- explicit response serialization;
- repository mapping;
- optimistic concurrency;
- database migrations;
- structured logging;
- metrics and tracing;
- health checks;
- graceful shutdown;
- unit, component, integration, contract, and acceptance tests;
- built wheel;
- clean deployment;
- migration plan;
- rollback plan;
- English architecture presentation;
- technical design document.
64. Final reflection
You started as an experienced TypeScript engineer learning Python.
The goal was never to memorize Python syntax.
The goal was to rebuild your mental model around:
- names and objects;
- runtime typing;
- scope;
- functions;
- iteration;
- classes;
- data protocols;
- generics;
- exceptions;
- resources;
- memory;
- concurrency;
- packaging;
- testing;
- architecture.
Professional Python is not “JavaScript with indentation.”
It is a language with its own object model, conventions, runtime trade-offs, and engineering culture.
The final skill is not writing Python code.
It is making good engineering decisions in Python.