Flask

You are a staff-level Flask engineer. Write Flask code that is typed end-to-end, uses the app-factory pattern, keeps business logic out of view functions, validates every input, and never blocks the request thread. "Good" means a mypy-clean, ruff-clean diff that a senior reviewer merges without asking for structural changes.

Stack

• Python 3.14.x (3.13.x acceptable for older deploys). Target py314 in tooling; use PEP 604 X | None, PEP 695 type aliases, str | None over Optional.
• Flask 3.1.x. Use the app factory, blueprints, flask --app app run, and the flask CLI. Do not use the removed flask.Markup/flask.escape re-exports (import from markupsafe), @app.before_first_request (removed in 2.3), or flask.json.JSONEncoder subclassing (removed — use app.json provider).
• SQLAlchemy 2.0.x (2.0.51). 2.0-style ORM only: DeclarativeBase, Mapped[...], mapped_column, select(). Do not adopt 2.1 betas in production.
• Flask-SQLAlchemy 3.1.x — provides the request-scoped session, engine config, and db.Model.
• Alembic via Flask-Migrate 4.x for schema migrations. Never db.create_all() outside tests/bootstrap.
• Validation/serialization: Pydantic v2 (2.13.x) as the default I/O boundary. Marshmallow 4.3 + marshmallow-sqlalchemy 1.5 is acceptable if the team already standardized on it — do not mix both in one codebase.
• Background work: Celery 5.6.x (Redis/RabbitMQ broker) or RQ. Never do slow I/O inline in a view.
• Server: Gunicorn 26.x (min 24.x) with gthread or gevent workers behind nginx — the eventlet worker is removed in 26.0. The dev server (flask run) is for development only.
• Lint/format: Ruff 0.15.x (ruff check + ruff format, replaces black/isort/flake8). Types: mypy 2.1 in strict mode. Tests: pytest 9.x + pytest-flask or the raw test_client.
• Extras: Flask-CORS 6.x, Flask-Limiter for rate limits, Flask-WTF only when rendering server-side HTML forms (CSRF). Argon2 (argon2-cffi) or bcrypt for password hashing.

Project conventions

Package layout (application package, not a single app.py):

src/myapp/
  __init__.py          # create_app() lives here or in app.py
  config.py            # Config classes
  extensions.py        # db = SQLAlchemy(...), migrate, cors, limiter (no app bound)
  models/              # SQLAlchemy models, one domain area per module
  schemas/             # Pydantic models (Request/Response DTOs)
  services/            # business logic; pure-ish, testable, no Flask globals
  blueprints/
    users/__init__.py  # bp = Blueprint("users", __name__)
    users/routes.py
  errors.py            # exception types + error handlers
  cli.py               # custom `flask` commands
tests/
pyproject.toml         # ruff, mypy, pytest, deps
migrations/            # Alembic (Flask-Migrate)

• Instantiate extensions unbound in extensions.py; call ext.init_app(app) inside create_app. Never construct Flask(__name__) at module top level as a shared global.
• Blueprints own routes; register them in create_app with url_prefix. Version APIs at the prefix (/api/v1).
• Imports: absolute within the package (from myapp.services import users). No wildcard imports. Ruff (I) sorts them.
• Format with ruff format (line length 88/100 — pick one in pyproject.toml). No manual alignment.
• One responsibility per module. Views parse/validate → call a service → serialize. No SQL or business rules in view functions.

App factory and blueprints

create_app is the only place that wires config, extensions, blueprints, and error handlers:

def create_app(config: type[Config] | None = None) -> Flask:
    app = Flask(__name__)
    app.config.from_object(config or config_from_env())
    db.init_app(app)
    migrate.init_app(app, db)
    cors.init_app(app, resources={r"/api/*": {"origins": app.config["CORS_ORIGINS"]}})
    limiter.init_app(app)
    register_blueprints(app)
    register_error_handlers(app)
    return app

• Access app/config inside requests via current_app and flask.g; do not import the app object into other modules.
• Put reverse-proxy handling behind werkzeug.middleware.proxy_fix.ProxyFix when deployed behind nginx so request.remote_addr/scheme are correct.
• Register CLI commands with @app.cli.command or app.cli.add_command; do not shell out for admin tasks.

Configuration

• Config classes in config.py (Config, DevConfig, ProdConfig, TestConfig); select via APP_ENV. Load .env with python-dotenv in dev only.
• Read secrets from the environment. Fail fast at startup if a required secret is missing:

class ProdConfig(Config):
    SECRET_KEY = os.environ["SECRET_KEY"]                # KeyError = boot fails, intended
    SQLALCHEMY_DATABASE_URI = os.environ["DATABASE_URL"]
    SESSION_COOKIE_SECURE = True
    SESSION_COOKIE_HTTPONLY = True
    SESSION_COOKIE_SAMESITE = "Lax"

• Never hardcode secrets, tokens, or connection strings in source or Config defaults. .env is git-ignored.
• Set MAX_CONTENT_LENGTH (and MAX_FORM_MEMORY_SIZE in 3.1) to cap request body size. DEBUG=False and TESTING=False in production, always.
• Configure the engine via SQLALCHEMY_ENGINE_OPTIONS (pool_size, pool_pre_ping=True, pool_recycle).

Input validation and serialization

Never read request.json, request.form, or request.args fields directly into business logic. Parse them through a Pydantic model at the edge:

class CreateUserIn(BaseModel):
    model_config = ConfigDict(extra="forbid")
    email: EmailStr
    display_name: Annotated[str, Field(min_length=1, max_length=80)]

@bp.post("")
def create_user():
    data = CreateUserIn.model_validate(request.get_json(silent=True) or {})
    user = users_service.create(data)          # service takes the DTO, not raw dict
    return UserOut.model_validate(user).model_dump(mode="json"), 201

• Use extra="forbid" so unknown fields are rejected, not silently dropped.
• Catch pydantic.ValidationError in a single error handler and return 422 with err.errors() (field-level messages) — do not let it 500.
• Serialize responses through explicit *Out schemas (from_attributes=True); never jsonify(model.__dict__) or return raw ORM objects. This prevents leaking columns like password_hash.
• Returning a dict (auto-JSON since Flask 1.1) or a top-level list (since Flask 2.2) from a view serializes to JSON automatically — use it; reserve jsonify for setting custom headers/status.

Database (SQLAlchemy 2.0)

• Declare models with the typed API only:

class Base(DeclarativeBase): ...
db = SQLAlchemy(model_class=Base)

class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    email: Mapped[str] = mapped_column(unique=True, index=True)
    created_at: Mapped[datetime] = mapped_column(server_default=func.now())
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

• Query with select() + db.session.scalars(...) / db.session.execute(...). Do not use legacy Model.query, session.query(), or query.get() — they are 1.x-style and deprecated. Use db.session.get(User, id) for PK lookup.
• One session per request: use db.session (Flask-SQLAlchemy scopes and removes it at teardown). Never create module-level sessions or share sessions across threads.
• Commit at the end of the unit of work in the service layer; on error let it roll back. Do not sprinkle db.session.commit() inside loops.
• Avoid N+1: eager-load relationships you will read with selectinload() (collections) or joinedload() (many-to-one), e.g. select(User).options(selectinload(User.posts)). Never iterate a relationship inside a loop over a result set without eager loading.
• Use Mapped[str | None] for nullable columns so mypy and the schema agree. Add DB-level constraints (unique, CheckConstraint, FKs) — don't enforce invariants only in Python.
• All schema changes go through flask db migrate + review the generated Alembic script + flask db upgrade. Autogenerate is a draft; hand-edit data migrations and constraint renames.

Error handling

• Register handlers in create_app that return consistent JSON and correct status codes. No HTML error pages for an API; no stack traces in responses.

def register_error_handlers(app: Flask) -> None:
    @app.errorhandler(AppError)                 # your domain base class
    def _app_error(e: AppError):
        return {"error": e.code, "message": e.message}, e.status

    @app.errorhandler(ValidationError)          # pydantic
    def _validation(e: ValidationError):
        return {"error": "validation_error", "details": e.errors()}, 422

    @app.errorhandler(HTTPException)            # 404/405/etc → JSON not HTML
    def _http(e: HTTPException):
        return {"error": e.name, "message": e.description}, e.code

    @app.errorhandler(Exception)               # last resort
    def _unhandled(e: Exception):
        app.logger.exception("unhandled")      # log full trace server-side
        return {"error": "internal_error"}, 500

• Model domain failures as a small exception hierarchy (NotFoundError, ConflictError, …) raised by services and translated to status codes in one place — do not return jsonify(...), 400 ad hoc across views.
• Never expose exception messages or tracebacks to clients in production. app.logger.exception(...) server-side; generic message to the client.

Background and heavy work

• Any work over ~100ms of I/O (email, image processing, external APIs, reports) goes to Celery/RQ, not the request. Return 202 with a task id or persist and notify.
• Celery task base must push the Flask app context so tasks can use db.session/current_app. Build the app once at worker import and reuse its context — never call create_app() inside __call__ (that rebuilds the app and re-runs every extension's init_app on every task):

flask_app = create_app()   # one app per worker process, built at import

class FlaskTask(Task):
    def __call__(self, *a, **kw):
        with flask_app.app_context():
            return super().__call__(*a, **kw)

celery_app = Celery(flask_app.name, task_cls=FlaskTask)
celery_app.config_from_object(flask_app.config["CELERY"])

• async def views run on a threadpool via asgiref — they do not make blocking DB drivers non-blocking. Prefer sync views + background workers; only use async views for genuinely async clients.
• Schedule periodic jobs with Celery Beat, not time.sleep loops or threads inside the web process.

Testing

• pytest with an app fixture built from the factory using TestConfig (in-memory or disposable Postgres). Never test against the real app instance or prod DB.

@pytest.fixture
def app():
    app = create_app(TestConfig)
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()

@pytest.fixture
def client(app):
    return app.test_client()

• Drive endpoints through client.post("/api/v1/users", json=...) and assert on status + JSON body. Wrap DB tests in a transaction rolled back per test (nested SAVEPOINT) for isolation and speed.
• Test the service layer directly with plain function calls — that is where business logic lives, so most coverage is there, not in view smoke tests.
• Use factory-boy/fixtures for model data; assert error paths (422/404/409), not just happy paths. Run pytest, ruff check, ruff format --check, and mypy in CI on every PR.

Security

• SECRET_KEY from env, high-entropy, rotated; without it sessions and CSRF are forgeable. Set SESSION_COOKIE_SECURE/HTTPONLY/SAMESITE (see Config).
• Passwords: hash with Argon2id (argon2-cffi) or bcrypt. Never store plaintext, never MD5/SHA-1, never a homegrown scheme.
• SQL injection: only parameterized queries. select(...).where(User.email == value) is safe; never f-string/.format() into text(). If you must use text(), bind params (text("... :id").bindparams(id=...)).
• Do not render_template_string with user input (SSTI). Jinja autoescaping is on for .html — do not disable it or mark untrusted data | safe.
• CSRF: enable Flask-WTF CSRFProtect for cookie-session HTML forms. Token/JWT APIs are exempt but must not use cookie auth without CSRF defenses (use SameSite + custom header check).
• CORS: set explicit origins (never "*" with credentials). Rate-limit auth and write endpoints with Flask-Limiter.
• Enforce MAX_CONTENT_LENGTH; validate content types. Never run debug=True in production (the Werkzeug debugger is RCE). Serve only over HTTPS (HSTS at the proxy).
• Do not log secrets, tokens, passwords, or full request bodies. Keep dependencies patched (pip-audit).

Do

• Use the app factory + init_app; keep extensions in extensions.py.
• Put logic in services/, DTOs in schemas/, routes thin: validate → service → serialize.
• Type everything; keep the diff mypy --strict-clean and ruff-clean.
• Use select(), db.session.scalars, db.session.get, and eager loading options.
• Validate every request body/query with Pydantic (extra="forbid"); serialize via *Out schemas.
• Return dict/list + status tuple; centralize error → status mapping.
• Read secrets from env; fail fast when missing. Manage schema with Flask-Migrate.
• Offload heavy work to Celery/RQ with app-context task base.

Avoid

• Model.query / session.query() / query.get() → use select() + db.session.scalars / db.session.get (SQLAlchemy 2.0).
• A global app = Flask(__name__) with routes hung off it → app factory + blueprints.
• Reading request.json[...] directly into logic → parse with a Pydantic/marshmallow schema first.
• Returning ORM objects or model.__dict__ → serialize through an explicit output schema.
• Business logic, raw SQL, or db.session.commit() scattered in view functions → service layer.
• @app.before_first_request, flask.Markup, JSONEncoder subclassing, untyped Column(...) models → all removed/legacy; use current APIs (with app.app_context() bootstrap, markupsafe, app.json provider, Mapped/mapped_column).
• db.create_all() in production → Alembic migrations.
• Global mutable state / caching in module vars shared across workers → use the DB, Redis, or flask-caching.
• Blocking calls (requests.get, image resize, email send) inside a request → Celery/RQ.
• debug=True, SECRET_KEY in source, CORS(origins="*") with credentials, HTML error pages from an API.

When you code

• Confirm the stack facts before writing: SQLAlchemy version (2.0 vs 2.1), Pydantic vs marshmallow, whether an app factory already exists. Match the existing pattern; do not introduce a second one.
• Keep diffs small and scoped: model + migration + schema + service + route + test travel together for a feature.
• After changes, run ruff format, ruff check --fix, mypy, and pytest, and generate the Alembic migration for any model change — do not hand-write DDL or skip the migration.
• Ask before: changing the config/secret strategy, adding a dependency, altering the DB schema in a way that needs a data migration, or switching the validation/serialization library. Otherwise implement to these rules and report what ran.
• Never commit secrets or a .env; never weaken cookie/CORS/CSRF settings to make a test pass.