From frontend to shipping backend

Full lesson lives in 01-request-lifecycle.html. Quick recap:

• Decorator = router. @app.get("/ping") registers the function below as the handler. Decorators are functions wrapping functions — read the body, not the name (see @development_only in routes.py, which only warns, doesn't disable).
• async runs on the event loop (same as JS). Blocking work there stalls every request on the worker → push it to a background task (like report_to_db).
• Return value = response body. Return a dict, get JSON. The -> OutputResponse annotation is the contract.
• CamelModel bridges camelCase JSON ⇄ snake_case Python.

Theory

In TS, types vanish at compile time — they never check real runtime data. Pydantic is the opposite: models validate and coerce incoming JSON at runtime. If the body doesn't match, FastAPI returns 422 automatically — you never write the validation.

two_picking_positions_engine/models/model.py

class InputRequest(CamelModel):
    order_id: int = Field(description="Order ID.")
    warehouse_id: int
    prioritization_time: datetime
    products: List[Products]
    country: Optional[str] = Field(None, ...)

Enums & Decimal: StorageType is an Enum (only MANUAL/AUTOMATED accepted). amount: Decimal — not float — because money/quantities need exactness.

Output is validated too: OutputResponse(products: Dict[int, str]) — a map of product id → sector. Return type -> OutputResponse shapes what leaves.

Pydantic: models · FastAPI: request body

Theory

There are 4 gunicorn worker processes. A Python variable in one worker is invisible to the others, and dies on restart. So shared, fast, live state lives in Redis — an in-memory key-value store outside the app.

This repo uses exactly two Redis structures per order (see redis_repository.py):

two_picking_positions_engine/repositories/redis_repository.py

# 1) HASH — full order detail, keyed by id
hash_key = f"order:{order.order_id}"
self.r.hset(hash_key, mapping={"data": order.json()})

# 2) SORTED SET — a priority queue: order_id scored by prio_time
sorted_set_key = f"warehouse:{warehouse_id}:orders"
self.r.zadd(sorted_set_key, {order_id: prioritization_time.timestamp()})

The hash answers "give me order 123's data". The sorted set answers "which orders come before this priority time?" (get_order_ids_before_prio_time → zrevrangebyscore). Score = unix timestamp, so time ordering is free.

Resilience: _execute_with_retries wraps every call with reconnect + exponential backoff. Batching: get_order_data_batch uses a pipeline to fetch many orders in one round-trip (a real perf pattern).

Redis: sorted sets · Redis: hashes · Redis University

Theory

Not every input is an HTTP request. Other systems publish events ("order PICKED", "order cancelled", "work volume recomputed") to RabbitMQ, and this service consumes them in the background to keep its Redis state current.

One consumer per country (per tenant vhost). It declares one queue bound to three exchanges (rabbit_consumer.py):

two_picking_positions_engine/consumers/rabbit_consumer.py

self.exchanges = [
  {"name":"wapi-events",       "type":"direct", "routing_key":"wapi-order-event"},
  {"name":"web.order.event",   "type":"topic",  "routing_key":"web.order.event.cancel"},
  {"name":"workers-balancing", "type":"topic",  "routing_key":"work-volume-prediction.computed"},
]

The callback routes by method.exchange: order events → update/remove orders in Redis; balancing events → update work-volume prediction. Then it acks the message (tells Rabbit "done, delete it").

RabbitMQ: topics & routing (Python/pika) · AMQP concepts: exchanges, queues, bindings

Theory

Redis is fast but ephemeral. For a permanent record of every decision (for analytics in Metabase), the service writes rows to PostgreSQL via the psycopg2 driver. This is the one place you write raw SQL.

two_picking_positions_engine/app/utilities.py · push_to_db()

connection = psycopg2.connect(host=mysql_secret['host'], ...)
cursor = connection.cursor()
insert_sql = f"""INSERT INTO {schema}.{table} (...) VALUES (%s, %s, ...)"""
cursor.executemany(insert_sql, data_to_insert)
connection.commit()

The values use %s placeholders — psycopg2 escapes them safely (this is how you avoid SQL injection). The table/schema names are f-string-interpolated, which is fine only because they're hard-coded constants.

Env gating: report_to_db returns early when SPRING_PROFILES_ACTIVE is development/testing — so local runs never touch the real analytics DB.

psycopg2: basic usage · SQLBolt (interactive SQL) · Why parameterise (SQL injection)

Theory

This is the biggest mental shift from the browser. In production the app runs under Gunicorn with 4 UvicornWorkers — four separate OS processes, each with its own memory, its own event loop, its own Redis connection.

two_picking_positions_engine/gunicorn_conf.py

worker_class = "uvicorn.workers.UvicornWorker"
workers = 4

Three kinds of concurrency live in this repo, and mixing them up is how you create bugs:

• Processes (gunicorn workers) — true parallelism, no shared memory → that's why state is in Redis.
• async / event loop — one thread, cooperative; great for I/O waiting, deadly if you block it.
• Threads — the repo runs the RabbitMQ consumer via asyncio.to_thread and a metrics monitor thread; BackgroundScheduler (APScheduler) runs update_redis hourly.

FastAPI: async, in a hurry · Gunicorn: design (workers) · FastAPI: background tasks

Theory

The app holds almost no config itself. At startup it pulls per-tenant settings (Redis host, RabbitMQ creds, DB) from a central Spring Cloud Config Server via the config-client library — selected by the SPRING_PROFILES_ACTIVE profile.

two_picking_positions_engine/configs/app_config.py

cfg_client = RhlConfigClient(
    address=os.getenv("RHL_CONFIG_SERVER", ...),
    app_name=f"_all_apps,{APP_NAME_DASHES}",
    profile=os.getenv("SPRING_PROFILES_ACTIVE", default="testing"),
    ...)
# later: get_app_config().get("app.tenants.cz.redis")

Multi-tenant: get_app_config().get(f"app.tenants.{country}") returns that country's Redis + RabbitMQ block — that's how one codebase serves at/cz/de/hu/ro. Secrets arrive encrypted and are decrypted client-side (encryption.py).

Theory

Everything converges here. To ship a change you need the loop: run → change → verify → commit → MR.

• Run: docker compose up (or poetry install + uvicorn ...app:app --reload). Hit localhost:8080/ping.
• Verify: curl your endpoint; use the dev routes; read the logs (TimingMiddleware).
• Ship: branch, commit (the repo's style is terse, lowercase, scoped), push, open a GitLab MR. CI runs Package → Tag → Deploy + an AI review. CODEOWNERS routes review to team-opex-outbound.

Poetry: basic usage · FastAPI: testing (TestClient)