FastAPI & Machine Learning: Build a Banking Fraud Detection

Install Python and pip, confirm versions, and set up Windows Subsystem for Linux. Create next gen bank fast API project with src directory, then create and activate Pipenv virtual environment.

Activate the virtual environment and install FastAPI with its standard dependencies using Pipenv, then install SQLModel, Alembic, psycopg3, asyncpg, Argon2, Pedantic, and Greenlet.

Install git and generate a python gitignore using toptal or npx, initialize a main branch, commit with vscode, and configure black formatter plus basic type checking in vscode.

Create a basic FastAPI backend by scaffolding a backend/app structure, defining a simple root route, and running a FastAPI dev server with automatic Swagger and Redoc documentation.

Organize folder structure by creating an api folder with __init__.py and main.py, plus a routes subfolder for home.py with /home route, and set up dot env templates and gitignore rule.

create a core config module in the app directory with a settings class. read local env files to supply api version, project name, and description.

Learn to use log guru for production-ready logging with file rotation, retention, and automatic exception tracebacks, plus seven levels and dynamic runtime control with colored console output.

Install log guru, configure a logging module, and set up rotating debug and error logs with a logs directory for a FastAPI app.

Learn to containerize an API with Docker by installing Docker Desktop and Docker Compose, then configure a Postgres image with a Dockerfile and Docker Compose.

Configure postgres environment variables and the database URL in a dockerized setup using env local and dot env files, with async pg and docker compose.

Explore how Traefik functions as a reverse proxy and load balancer to route requests, perform SSL termination, and auto-discover services across docker environments via hostname-based routing and health checks.

Create a static Traefik yaml config for development, enabling the api dashboard, insecure access, json logging, and a retry middleware with three attempts and 500 ms interval.

Create a startup script to run a FastAPI app with Uvicorn in Docker, configure robust shell options, expose on 0.0.0.0:8000 with reload, and prepare backend requirements for the Docker workflow.

Use a multi-stage dockerfile with python 3.13.1-slim-bookworm to build dependency wheels and install requirements, producing a lean development and production image.

Continue the dockerfile by setting logs ownership to the app user and group, installing dependencies from wheels with no cache, and preparing entrypoint and start scripts for a FastAPI container.

Configure dockerignore to exclude git and py caches, set up docker compose with traffic and api services, and integrate mail pit for development on the next gen local network.

Set up celery with Redis and RabbitMQ for asynchronous background tasks in FastAPI, with Flower for monitoring. Install and manage email tools with FastAPI-Mail, aio-smtplib, and email-validator using Pipenv.

Configure a celery app for the FastAPI banking project by creating salary_app.py, connecting a RabbitMQ broker and radius backend, and enabling JSON task and result serialization.

Configure docker-compose to wire api, celery, flower, redis, and rabbitmq, with append-only redis, traffic routing, shared networks, and persistent volumes.

Configure and rebuild docker containers with a local docker-compose file, fix the flower error, verify routes and dashboards, and introduce a race condition to be addressed in the next lesson.

Diagnose container race conditions across traffic api, fast api, postgres, rabbitmq, and celery workers and implement health checks, dependency ordering, and retry logic to fail fast and avoid startup issues.

Implement health checks, dependency validation, and retry logic to prevent race conditions, ensuring the application only starts accepting requests when all services are healthy and ready.

Extend health checks by adding an async add service function to register services with timeout, retries, and dependencies, and implement a database health check using an async session.

Check the health of all registered services by calling each service's health function, using cache to avoid redundant checks, and returning a health status dictionary with timestamps.

Implement an asynchronous health check that waits for all registered services to become healthy within a 30-second timeout, polling status, cleaning up resources on shutdown, and instantiating the health checker.

Refactor db.py to implement async adapted pool, health checks, and a FastAPI session factory with dependency injection, plus retry with exponential backoff and connection verification.

Configure OTP settings and login security in the core config, then scaffold an auth module with a user schema, including security questions, account status, and role enums for banking API.

Define the user model for the database by inheriting from the base user schema, with uuid primary key, hashed password, OTP fields, timezone timestamps, and a computed full name.

Create a model registry that automatically discovers and imports all models at application startup, enabling sustainable, maintainable model loading for FastAPI banking with AI/ML fraud detection.

Initialize Alembic for async migrations with alembic init -t async migrations, configure env.py to load models, set database URL, set target metadata to SQLModel.metadata, and update script.py to import SQLModel.

Rebuild containers with Docker Compose -f local.yml, verify the model registry loads the user model, and run Alembic to add the user table and apply migrations.

Create a backend auth utils.py to generate a six-digit otp, hash and verify passwords with Argon2, and auto-generate bank usernames from site name with a prefix and hyphenated random characters.

Create a base email template class using the Jinja environment to render HTML and plain text emails from templates with a context, and send via a celery task.

Create a base html email template using jinja inheritance with header, content, and footer blocks, plus a plain text version, enabling child templates to override sections for consistent styling.

Configure activation token expiration, API base URL, support email, and JWT settings in config.py; generate a secure JWT secret with secrets, install pyjwt, and extend user read and login schemas.

Create an activation email service using the existing template to send activation emails with a generated activation URL and a JWT token with configured expiry.

Implement async session-based user retrieval by id or email, check existence, verify passwords with utils, and reset user state (login attempts, otp, and account status) with commit.

creates a login OTP email service by wiring templates and settings for OTP, expiry, and support email, with retry logic and exponential backoff for three attempts.

Verify login OTP enforces valid, non-expired codes, tracks failed attempts, and manages account lockout by resetting user state and guiding users to request a new OTP.

Create the register route in a FastAPI auth module, validating email and ID, handling errors, logging, and returning an HTTP 201 created user via the user read schema.

Post requests to the recent activation link route validate email, check user existence and activation status, generate a new activation token, and send the activation email.

Group and expose auth routes in the main api router by including register and activate routers. Test with postman, validate jwt-based authentication, and observe activation status and token expiry.

Implement cookie-based authentication in fastapi using three cookies: access, refresh, and login tokens, configuring their lifetimes and setting and deleting cookies with proper security settings.

Implement the verify login OTP route at post /login/verify-otp to validate email and OTP, manage the session, create tokens, set cookies, and update main.py to include the login router.

Demonstrate the login OTP workflow by requesting and verifying one-time passwords in Postman, observe account lock after three failed attempts, and review OTP expiry along with auth cookies and tokens.

Create account lockout email templates in html and txt, detailing account security alert, lockout duration, and post-unlock steps, including password change, contact support, and checks for suspicious activity.

create an account lockout email service that sends security alert messages using the account lockout template, computing unlock time from settings and formatting timestamps with site name and support details.

Refactor the user auth service to increment failed login attempts, update the last failed login timestamp in UTC, and send an account lockout email when the limit is reached.

Create password reset schemas with pydantic email validation, enforce password length and ensure passwords match, and implement time-bound JWT tokens for resets.

Implement a password reset email service in a FastAPI backend, including token generation, reset URL creation, and secure password updates validated by JWT with error handling.

Implement a token refresh endpoint in a FastAPI banking app to issue new access tokens from valid refresh tokens, using jwt validation, cookies, and user authentication services.

Integrate the refresh route into main.py, test JWT refresh via postman using OTP and cookies, and refactor token expiration to use days for refresh tokens.

Implement a logout endpoint in FastAPI to delete auth cookies, log the user out, and handle errors, updating main.py and testing with Postman to confirm cookies are cleared.

Configure Cloudinary for user profile images and documents, set up API keys and environment variables, install media handling packages, and prepare Alembic migrations to run before startup.

Build the profiles model separate from the users model to handle know your customer data, and define a profile base schema using SQLModel and Pydantic to validate KYC fields.

Define a profile model with a one-to-one relation to the user using SQLModel, including createdat, updatedat, timezone, and a user_id foreign key, plus bidirectional access via profile.user and user.profile.

Define the profile create schema by subclassing the base schema and using a Pydantic field validator to validate ID issue and expiry dates with a utility function.

Test the profile creation route on Postman after login with OTP, sending a payload with title, gender, and date of birth, and verify the user foreign key.

Update the profile update schema by subclassing the profile base schema with all fields optional and defaulting to none, and add an image type schema and ID expiry validator.

Develop the profile update service by retrieving the user profile, updating allowed fields from the update schema, excluding unset and photo fields, with proper logging and error handling.

Define a profile update route in FastAPI using a patch request to /profile/update, wiring update_user_profile with the profile update schema, current user, and session, while enabling logging and error handling.

Add and test a FastAPI profile update route by sending a patch request to update title, marital status, means of identification, and employer name; verify 200 OK with refresh token.

Refactor celery tasks to a centralized folder, update imports and autodiscover paths, and configure cloudinary settings for images (mime types, max size 25MB, max dimension 4096) for streamlined uploads.

Develop and test the profile image upload task for KYC, enabling users to upload profile, ID, and signature images with mime type and size validations via celery and cloudinary.

Pre-upload image verification uses the Pillow library to validate size up to five megabytes, formats JPEG or PNG, and dimensions against settings.

This lecture introduces a two-step profile image upload service in FastAPI, using Celery for asynchronous uploads and updating the user profile with the image URL (profile, id, or signature photo).

Create a profile image upload route in fastapi that updates the user’s profile image url. Validate the image and schedule a background upload with celery, returning a task id.

Implement a get upload status route to query a background profile image upload by task ID, returning 200 when ready and updating the user profile with the image URL.

Test the image upload process by adding the upload route to main.py and uploading id, profile, and signature photos via Postman, tracking tasks and Cloudinary URLs.

This lecture guides building the profile response schema and the get user with profile service to fetch a user's profile via route, including username, names, email, id number, and role.

Create a paginated profile response schema and an async get all user profiles endpoint with branch manager permission checks, skip/limit pagination, ordered by createdat, and returning total.

Add the all profiles router to the main router, test with postman, verify role-based access by elevating a user to branch manager to fetch all profiles with skip and limit.