:::warning AUTO-GENERATED — do not edit This page is generated from the MCP server snapshot content/backend-mcp.json. Edit the source MCP server (not this file), then run npm run generate. :::

Backend — Templates & Patterns

Canonical, verified code templates and patterns for backend endpoints.

Endpoint template

Flask blueprint route with pydantic request/response schema validation, limit/offset pagination, and a standard success/error envelope. Every route is guarded by @require_auth + @require_role.

{
  "id": "endpoint",
  "description": "Flask blueprint route with pydantic request/response schema validation, limit/offset pagination, and a standard success/error envelope. Every route is guarded by @require_auth + @require_role.",
  "deps": [
    "flask",
    "pydantic>=2"
  ],
  "code": "from typing import Any\n\nfrom flask import Blueprint, request, g, jsonify\nfrom pydantic import BaseModel, Field, ValidationError\n\n# Auth decorators come from the auth pattern (see get_auth_pattern).\nfrom .auth import require_auth, require_role\n# Structured logger comes from the logging pattern (see get_logging_pattern).\nfrom .logging_config import logger\n\nwells_bp = Blueprint(\"wells\", __name__, url_prefix=\"/api/wells\")\n\n\n# ---- Schemas -----------------------------------------------------------------\nclass ListWellsQuery(BaseModel):\n    \"\"\"Validated query params, including pagination bounds.\"\"\"\n\n    operator: str = Field(min_length=1)\n    limit: int = Field(default=50, ge=1, le=200)\n    offset: int = Field(default=0, ge=0)\n\n\nclass WellResponse(BaseModel):\n    well_id: int\n    name: str\n    operator: str\n\n\n# ---- Standard envelope -------------------------------------------------------\ndef success_response(data: Any, meta: dict[str, Any] | None = None):\n    return jsonify({\"data\": data, \"meta\": meta or {}}), 200\n\n\ndef error_response(status: int, message: str, details: Any = None):\n    return (\n        jsonify({\"error\": {\"code\": status, \"message\": message, \"details\": details}}),\n        status,\n    )\n\n\n# ---- Route -------------------------------------------------------------------\n@wells_bp.get(\"\")\n@require_auth\n@require_role(\"wells:read\")\ndef list_wells():\n    try:\n        query = ListWellsQuery(**request.args.to_dict())\n    except ValidationError as err:\n        return error_response(422, \"Invalid query parameters\", err.errors())\n\n    logger.info(\n        {\n            \"event\": \"list_wells\",\n            \"user_id\": g.user_id,\n            \"operator\": query.operator,\n            \"limit\": query.limit,\n            \"offset\": query.offset,\n        }\n    )\n\n    # Replace with the db-read helper (see get_db_pattern). It should apply\n    # limit/offset in SQL; sliced here only to illustrate the envelope.\n    rows: list[dict[str, Any]] = []\n    page = rows[query.offset : query.offset + query.limit]\n\n    payload = [WellResponse(**row).model_dump() for row in page]\n    meta = {\"limit\": query.limit, \"offset\": query.offset, \"total\": len(rows)}\n    return success_response(payload, meta)\n",
  "notes": [
    "Register the blueprint with app.register_blueprint(wells_bp); do not define routes on the bare app.",
    "Both decorators are required: @require_auth then @require_role — order matters (auth runs first as the outer/nearest-to-def wrapper resolves bottom-up).",
    "Pagination is part of the envelope: always return meta.limit/offset/total so clients can page deterministically.",
    "Never return raw exceptions to clients; funnel everything through error_response so the envelope shape is stable."
  ]
}

Auth pattern

JWKS client + @require_auth decorator (validates signature/issuer/expiry, sets g.user_id=re_id and g.roles=role) + @require_role. Zero-trust: /health is the only public allowlist entry.

{
  "id": "auth",
  "description": "JWKS client + @require_auth decorator (validates signature/issuer/expiry, sets g.user_id=re_id and g.roles=role) + @require_role. Zero-trust: /health is the only public allowlist entry.",
  "deps": [
    "flask",
    "pyjwt[crypto]>=2"
  ],
  "code": "from functools import wraps\n\nimport jwt\nfrom jwt import PyJWKClient\nfrom flask import request, g, jsonify\n\n# Configure from environment/secrets in real code; shown inline for clarity.\nJWKS_URL = \"https://login.rystadenergy.com/.well-known/jwks.json\"\nISSUER = \"https://login.rystadenergy.com/\"\nAUDIENCE = \"rystad-dashboards\"\n\n# Zero-trust: every route requires auth EXCEPT this explicit allowlist.\nPUBLIC_PATHS = frozenset({\"/health\"})\n\n# Reused across requests; caches the JWKS and refreshes on rotation.\n_jwks_client = PyJWKClient(JWKS_URL)\n\n\ndef _auth_error(status: int, message: str):\n    response = jsonify({\"error\": {\"code\": status, \"message\": message}})\n    response.status_code = status\n    return response\n\n\ndef require_auth(fn):\n    \"\"\"Validate the bearer token's signature, issuer, audience and expiry.\"\"\"\n\n    @wraps(fn)\n    def wrapper(*args, **kwargs):\n        if request.path in PUBLIC_PATHS:\n            return fn(*args, **kwargs)\n\n        auth_header = request.headers.get(\"Authorization\", \"\")\n        if not auth_header.startswith(\"Bearer \"):\n            return _auth_error(401, \"Missing or malformed Authorization header\")\n\n        token = auth_header.split(\" \", 1)[1]\n        try:\n            signing_key = _jwks_client.get_signing_key_from_jwt(token)\n            claims = jwt.decode(\n                token,\n                signing_key.key,\n                algorithms=[\"RS256\"],\n                audience=AUDIENCE,\n                issuer=ISSUER,\n                options={\"require\": [\"exp\", \"iss\", \"sub\"]},\n            )\n        except jwt.InvalidTokenError as err:\n            return _auth_error(401, f\"Invalid token: {err}\")\n\n        re_id = claims.get(\"re_id\")\n        if re_id is None:\n            return _auth_error(401, \"Token is missing the re_id claim\")\n\n        role = claims.get(\"role\", [])\n        g.user_id = re_id\n        g.roles = [role] if isinstance(role, str) else list(role)\n        return fn(*args, **kwargs)\n\n    return wrapper\n\n\ndef require_role(*required_roles: str):\n    \"\"\"Authorize the request against g.roles set by require_auth.\"\"\"\n\n    def decorator(fn):\n        @wraps(fn)\n        def wrapper(*args, **kwargs):\n            roles = getattr(g, \"roles\", None) or []\n            if not any(role in roles for role in required_roles):\n                return _auth_error(403, \"Insufficient role for this resource\")\n            return fn(*args, **kwargs)\n\n        return wrapper\n\n    return decorator\n",
  "notes": [
    "Signature is verified against the JWKS public key — never disable verify_signature and never accept HS256/none.",
    "iss, aud and exp are all enforced; options.require guarantees exp/iss/sub are present.",
    "require_role reads g.roles, so it MUST be applied inside (below) require_auth on the handler.",
    "PUBLIC_PATHS is the single source of truth for anonymous access — add nothing but /health without a security review."
  ]
}

Database pattern

Read/write DB access through dynamixWrapper.dbconnect. Reads use get_sqlalchemy_connection_string(driver, server, db) + SQLAlchemy text() parameterized SELECT into a typed result; writes use update_table(userid, data, table, driver, server, db, columns=...).

{
  "id": "db",
  "description": "Read/write DB access through dynamixWrapper.dbconnect. Reads use get_sqlalchemy_connection_string(driver, server, db) + SQLAlchemy text() parameterized SELECT into a typed result; writes use update_table(userid, data, table, driver, server, db, columns=...).",
  "deps": [
    "sqlalchemy>=2",
    "pyodbc",
    "dynamixWrapper (internal package: 'Dynamic Algo Wrapper/dynamixWrapper')"
  ],
  "code": "from typing import TypedDict\n\nfrom sqlalchemy import create_engine, text\n\n# Real wrapper package — do NOT hand-roll connection strings or raw drivers.\nfrom dynamixWrapper.dbconnect import (\n    get_sqlalchemy_connection_string,\n    update_table,\n)\n\n\nclass WellRow(TypedDict):\n    well_id: int\n    name: str\n    operator: str\n\n\n# ---- READ: parameterized SELECT into a typed result --------------------------\ndef read_wells(\n    driver: str,\n    server: str,\n    db: str,\n    operator: str,\n    limit: int = 50,\n    offset: int = 0,\n) -> list[WellRow]:\n    connection_string = get_sqlalchemy_connection_string(driver, server, db)\n    engine = create_engine(connection_string, pool_pre_ping=True)\n\n    query = text(\n        \"SELECT well_id, name, operator \"\n        \"FROM dbo.wells \"\n        \"WHERE operator = :operator \"\n        \"ORDER BY well_id \"\n        \"OFFSET :offset ROWS FETCH NEXT :limit ROWS ONLY\"\n    )\n\n    with engine.connect() as connection:\n        result = connection.execute(\n            query,\n            {\"operator\": operator, \"limit\": limit, \"offset\": offset},\n        )\n        return [\n            WellRow(well_id=row.well_id, name=row.name, operator=row.operator)\n            for row in result\n        ]\n\n\n# ---- WRITE: delegate to the wrapper's retrying update_table -------------------\ndef write_wells(\n    userid: int,\n    rows: list[list[object]],\n    driver: str,\n    server: str,\n    db: str,\n) -> None:\n    # 'rows' is a 2D list; each inner list aligns with 'columns'.\n    # update_table clears the user's existing rows and re-inserts (fk_user\n    # is added automatically), with exponential-backoff retries built in.\n    update_table(\n        userid,\n        rows,\n        \"wells\",\n        driver,\n        server,\n        db,\n        columns=[\"name\", \"operator\", \"spud_date\"],\n    )\n",
  "notes": [
    "get_sqlalchemy_connection_string(driver, server, db) is the ONLY sanctioned way to build an engine URL — no raw mssql+pyodbc strings in app code.",
    "All SQL uses text() with bound :params — never string-format user input into SQL.",
    "update_table's real signature is (userid, data, table, driver, server, db, columns=None, ...); driver/server/db are required positionals, not optional.",
    "update_table is destructive per user: it DELETEs where fk_user = userid before inserting, and injects fk_user for you — do not include fk_user in columns/data.",
    "Reads are read-only SELECTs; do not issue INSERT/UPDATE/DELETE through the read engine."
  ]
}

Logging pattern

Structured logging configured once at application startup via python_logger.setup_logger(frame, session_guid, format, log_level, app_name). The returned logger is imported everywhere instead of print().

{
  "id": "logging",
  "description": "Structured logging configured once at application startup via python_logger.setup_logger(frame, session_guid, format, log_level, app_name). The returned logger is imported everywhere instead of print().",
  "deps": [
    "python_logger (internal package: 'pythonLogger/python_logger')"
  ],
  "code": "import inspect\nimport logging\n\n# Real logging package — configures file/Elasticsearch handlers + JSON output.\nfrom python_logger.SetupLogging import setup_logger\n\n# Configure ONCE at startup (app factory / wsgi entrypoint). setup_logger reads\n# frame[0] to derive the caller module, so pass inspect.stack()[0].\nlogger = setup_logger(\n    inspect.stack()[0],\n    session_guid=None,\n    log_level=logging.INFO,\n    app_name=\"rystad-backend\",\n)\n\n# Emit structured events (dict messages become JSON fields) — never print().\nlogger.info({\"event\": \"startup\", \"service\": \"rystad-backend\"})\n\n\ndef get_logger():\n    \"\"\"Import this in modules that need the shared, configured logger.\"\"\"\n    return logger\n",
  "notes": [
    "setup_logger's first arg is a stack FrameInfo (it indexes frame[0]); pass inspect.stack()[0], not a module or string.",
    "Signature is setup_logger(frame, session_guid=None, format={}, log_level=logging.INFO, app_name='dynamixWrapper') — set app_name to your service.",
    "Call setup_logger exactly once at startup; share the returned logger (e.g. via get_logger) rather than reconfiguring per module.",
    "Pass dict messages (logger.info({...})) to get structured JSON fields; plain strings are logged as the 'message' field.",
    "print() is banned by the guardrails — always route output through this logger."
  ]
}

Endpoint template​

Auth pattern​

Database pattern​

Logging pattern​

Endpoint template

Auth pattern

Database pattern

Logging pattern