Day 3: FastAPI Scaffold & State Machine

• ~5 min read

What I Built

  • FastAPI application with async MySQL 8.4 + asyncmy driver
  • Complete SQLAlchemy ORM models adapted for MySQL (VARCHAR UUIDs)
  • Account state machine with Redis-based distributed locking
  • Comprehensive unit tests (18 tests, all passing)
  • Health check endpoint and proper async session management

Code Highlight

# FastAPI app with health check
app = FastAPI(
    title="ZephyrApex",
    description="AI-augmented trading engine",
    version="0.1.0",
    lifespan=lifespan
)

# Health check endpoint
@app.get("/api/health")
async def health():
    return {"status": "ok", "version": "0.1.0"}

# State machine with Redis locking
class AccountStateMachine:
    def __init__(self, session: AsyncSession, lock_manager: LockManager):
        self.session = session
        self.lock_manager = lock_manager

    async def transition(self, account_id: str, from_state: AccountState, to_state: AccountState):
        async with self.lock_manager.account_lock(account_id):
            # Validate transition logic
            # Update with optimistic locking
            # Commit transaction
            pass

Architecture Decision

Chose MySQL VARCHAR(36) for UUIDs instead of PostgreSQL's native UUID type. Trade-off: storage efficiency vs familiarity. The state machine uses Redis context managers for distributed locking instead of PostgreSQL advisory locks - simpler, more scalable, but requires Redis dependency.

Testing Results

All 18 unit tests pass, covering critical scenarios:

  • Valid and invalid state transitions
  • Lock acquisition and release
  • Account not found error handling
  • Optimistic locking conflict detection
  • Exception handling with proper lock cleanup
  • Concurrent access prevention

Next Steps

Day 4: Write-Ahead Log implementation for order management and reconciliation logic.

Follow @therealkamba on X for regular updates. View all posts →