This package provides a set of utilities to help you build ASGI applications with Python.
Stack that we consider that the project uses:
The repository provides a Cookiecutter template for
bootstrapping a new FastAPI application with c2casgiutils:
pip install cookiecutter
cookiecutter https://github.com/camptocamp/c2casgiutils --directory scaffoldCookiecutter will prompt for the project name (project_slug) and description, then generate a
ready-to-use project directory. Start the application with:
cd <project_slug>
docker compose up --buildSee scaffold/README.md for the full setup guide.
The scaffold Dockerfile uses uvicorn to start the application. Typical flags include:
--host=0.0.0.0to bind on all interfaces inside the container.--port=8080to expose the HTTP port.--log-config=/app/logging.yamlto load the logging configuration.
See the Uvicorn settings reference for the full list of options: https://www.uvicorn.org/settings/
When the app runs behind a reverse proxy (Kubernetes Ingress, Traefik, nginx, etc.) you should trust forwarded headers.
If the Host header has the right values you can use the option provided by Uvicorn:
--proxy-headers --forwarded-allow-ips=*--proxy-headersmakes Uvicorn trustX-Forwarded-Proto,X-Forwarded-For, and related headers.--forwarded-allow-ips=*allows forwarded headers from any upstream proxy. If you know your proxy IPs, prefer a strict list instead of*to harden the configuration.
If the Host header is not correct, for example with Apache and the default configuration,
the header X-Forwarded-Host or Forwarded should also be interpreted.
In that case, ForwardedHeadersMiddleware is required.
To use the RFC7239 Forwarded header, set C2C__PROXY_HEADERS__TYPE=forwarded.
Or use C2C__PROXY_HEADERS__TYPE=x-forwarded to trust X-Forwarded-*.
Use C2C__PROXY_HEADERS__TRUSTED_HOSTS=... to restrict which proxy IPs are trusted (use * only if you must).
Note: when --proxy-headers is enabled, Uvicorn updates the client address before
ForwardedHeadersMiddleware runs. That means trusted_hosts is matched against the updated
client address, not the direct peer connection. If you want this middleware to validate the
direct proxy IP, run Uvicorn without --proxy-headers and rely on the middleware instead.
pip install c2casgiutils[all]Add in your application:
import c2casgiutils
from c2casgiutils import broadcast
from c2casgiutils import config
from prometheus_client import start_http_server
from prometheus_fastapi_instrumentator import Instrumentator
from contextlib import asynccontextmanager
@asynccontextmanager
async def _lifespan(main_app: FastAPI) -> None:
"""Handle application lifespan events."""
_LOGGER.info("Starting the application")
await c2casgiutils.startup(main_app)
yield
app = FastAPI(title="My fastapi_app application", lifespan=_lifespan)
app.mount('/c2c', c2casgiutils.app)
# For security headers (and compression)
# Add TrustedHostMiddleware (should be first)
app.add_middleware(
TrustedHostMiddleware,
allowed_hosts=["*"], # Configure with specific hosts in production
)
# Add GZipMiddleware
app.add_middleware(GZipMiddleware, minimum_size=1000)
# Set all CORS origins enabled
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.add_middleware(headers.ArmorHeaderMiddleware,
headers_config={
"http": {"headers": {"Strict-Transport-Security": None} if not config.settings.http else {}},
}
)
# Optional: trust host/port from forwarded proxy headers
if config.settings.proxy_headers.type != "none":
app.add_middleware(
headers.ForwardedHeadersMiddleware,
trusted_hosts=config.settings.proxy_headers.trusted_hosts,
headers_type=config.settings.proxy_headers.type,
)
# Get Prometheus HTTP server port from environment variable 9000 by default
start_http_server(config.settings.prometheus.port)
instrumentator = Instrumentator(should_instrument_requests_inprogress=True)
instrumentator.instrument(app)To use the broadcasting you should do something like this:
import c2casgiutils
from c2casgiutils.broadcast import MissingAnswer
from c2casgiutils.broadcast import types as broadcast_types
from typing import Protocol
class EchoResponse(BaseModel):
"""Response from broadcast endpoint."""
result: list[dict[str, Any]] | None = None
class EchoHandlerProto(Protocol):
async def __call__(self, *, message: str) -> list[broadcast_types.BroadcastResponse[EchoResponse]|MissingAnswer] | None: ...
# Late assignment
echo_handler: EchoHandlerProto = None # type: ignore[assignment]
# Create a handler that will receive broadcasts
async def _echo_handler(*, message: str) -> EchoResponse:
"""Echo handler for broadcast messages."""
return EchoResponse(message="Broadcast echo: " + message)
# Subscribe the handler to a channel on module import
@asynccontextmanager
async def _lifespan(main_app: FastAPI) -> None:
"""Handle application lifespan events."""
_LOGGER.info("Starting the application")
await c2casgiutils.startup(main_app)
# Register the echo handler
global echo_handler # pylint: disable=global-statement
echo_handler = await broadcast.decorate(_echo_handler, expect_answers=True)
yieldThen you can use the echo_handler function you will have the response of all the registered applications.
The health_checks module provides a flexible system for checking the health of various components of your application. Health checks are exposed through a REST API endpoint at /c2c/health_checks and are also integrated with Prometheus metrics.
To initialize health checks in your application:
from c2casgiutils import health_checks
# Add Redis health check
health_checks.FACTORY.add(health_checks.Redis(tags=["liveness", "redis", "all"]))
# Add SQLAlchemy database connection check
from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
health_checks.FACTORY.add(health_checks.SQLAlchemy(Session=your_async_sessionmaker, tags=["database", "all"]))
# Add Alembic migration version check
health_checks.FACTORY.add(health_checks.Alembic(
Session=your_async_sessionmaker,
config_file="alembic.ini",
tags=["migrations", "database", "all"]
))The package provides several built-in health check implementations:
- Redis: Checks connectivity to Redis by pinging both master and slave instances
- SQLAlchemy: Verifies database connectivity by executing a simple query
- Alembic: Ensures the database schema is up-to-date with the latest migrations
You can create custom health checks by extending the Check base class:
from c2casgiutils.health_checks import Check, Result
class MyCustomCheck(Check):
async def check(self) -> Result:
# Your check logic here
try:
# Perform your check...
return Result(status_code=200, payload={"message": "Everything is fine!"})
except Exception as e:
return Result(status_code=500, payload={"error": str(e)})
# Add your custom check
health_checks.FACTORY.add(MyCustomCheck(tags=["custom", "all"]))Health checks can be filtered using tags or names:
- Tags: Add relevant tags when creating a check to categorize it
- API Filtering: Use query parameters to filter checks when calling the API:
/c2c/health_checks?tags=database,critical- Run only checks with "database" or "critical" tags/c2c/health_checks?name=Redis- Run only the Redis check
Health check results are automatically exported to Prometheus metrics via the health_checks_failure gauge, allowing you to monitor and alert on health check failures.
The ArmorHeaderMiddleware provides automatic security headers configuration for your ASGI application. It allows you to configure headers based on request netloc (host:port) and path patterns.
from c2casgiutils.headers import ArmorHeaderMiddleware
# Use default security headers
app.add_middleware(ArmorHeaderMiddleware)
# Or with custom configuration
app.add_middleware(ArmorHeaderMiddleware, headers_config=your_custom_config)The middleware comes with sensible security defaults including:
- Content-Security-Policy: Restricts resource loading to prevent XSS attacks
- X-Frame-Options: Prevents clickjacking by denying iframe embedding
- Strict-Transport-Security: Forces HTTPS connections (disabled for localhost)
- X-Content-Type-Options: Prevents MIME-type sniffing
- Referrer-Policy: Controls referrer information sent with requests
- Permissions-Policy: Restricts access to browser features like geolocation
- X-DNS-Prefetch-Control: Disables DNS prefetching
- Expect-CT: Certificate Transparency enforcement
- Origin-Agent-Cluster: Isolates origin agent clusters
- Cross-Origin policies: CORP, COOP, COEP for cross-origin protection
You can configure headers based on request patterns:
from c2casgiutils.headers import ArmorHeaderMiddleware
headers_config = {
"api_endpoints": {
"path_match": r"^/api/.*", # Regex pattern for paths
"headers": {
"Access-Control-Allow-Origin": "*",
"X-Custom-Header": "api-value"
},
"order": 1 # Processing order
},
"admin_section": {
"netloc_match": r"^admin\..*", # Regex for host matching
"path_match": r"^/admin/.*",
"headers": {
"X-Robots-Tag": "noindex, nofollow"
},
"status_code": 200, # Only apply for specific status code
"order": 2
},
"success_responses": {
"headers": {
"Cache-Control": ["public", "max-age=3600"]
},
"status_code": (200, 299), # Apply for a range of status codes (200-299)
"order": 3
},
"api_methods": {
"path_match": r"^/api/.*",
"methods": ["GET", "HEAD"], # Only apply for specific HTTP methods
"headers": {
"Cache-Control": ["public", "max-age=3600"]
},
"order": 4
},
"remove_header": {
"headers": {
"Server": None # Remove header by setting to None
}
}
}
app.add_middleware(ArmorHeaderMiddleware, headers_config=headers_config)Headers support multiple value types:
headers_config = {
# String value
"X-Custom": "value",
# List (joined with "; ")
"Cache-Control": ["no-cache", "no-store", "must-revalidate"],
# Dictionary (for complex headers like CSP)
"Content-Security-Policy": {
"default-src": ["'self'"],
"script-src": ["'self'", "https://cdn.example.com"],
"style-src": ["'self'", "'unsafe-inline'"]
},
# List (joined with ", ") for Permissions-Policy
"Permissions-Policy": ["geolocation=()", "microphone=()"],
# Remove header
"Unwanted-Header": None
}For dynamic content that requires inline scripts or styles, using nonces is more secure than 'unsafe-inline'. A nonce is a random value that must match between the CSP header and the script/style tags.
from c2casgiutils import headers
# Configure CSP with nonce
custom_config = {
"my_page": {
"path_match": r"^/my-page/?",
"headers": {
"Content-Security-Policy": {
"default-src": ["'self'"],
"script-src": ["'self'", headers.CSP_NONCE],
"style-src": ["'self'", headers.CSP_NONCE],
}
}
}
}In your endpoint, pass the nonce to the template
from fastapi.templating import Jinja2Templates
# Configure templates
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
return templates.TemplateResponse(
"index.html.jinja2",
{
"request": request,
"nonce": getattr(request.state, "nonce", ""),
},
)In your HTML template, add the nonce attribute
<script nonce="{{ nonce }}">
...
</script>
<style nonce="{{ nonce }}">
...
</style>The middleware automatically disables Strict-Transport-Security for localhost to facilitate development.
You can apply headers conditionally based on response status codes:
- Apply to a single status code:
"status_code": 200 - Apply to a range of status codes:
"status_code": (200, 299)(for all 2xx success responses)
This feature is useful for adding caching headers only to successful responses, or special headers for specific error codes.
You can configure headers to be applied only for specific HTTP methods:
{
"api_post_endpoints": {
"path_match": r"^/api/.*",
"methods": ["POST", "PUT", "PATCH"], # Only apply for these methods
"headers": {
"Cache-Control": "no-store"
}
},
"api_get_endpoints": {
"path_match": r"^/api/.*",
"methods": ["GET", "HEAD"], # Only apply for GET and HEAD requests
"headers": {
"Cache-Control": ["public", "max-age=3600"]
}
}
}This allows for fine-grained control over which headers are applied based on the request method, useful for implementing different caching strategies for read vs. write operations.
With the default CSP your html application will not work, to make it working without impacting the security Of the other pages you should add in the headers_config something like this:
from c2casgiutils import headers
{
"my_page": {
"path_match": r"^your-path/?",
"headers": {
"Content-Security-Policy": {
"default-src": ["'self'"],
"script-src-elem": ["'self'", ...],
"style-src-elem": ["'self'", ...],
}
},
"order": 1
}
}And do the same for other headers.
The Cache-Control header can be configured to control caching behavior for different endpoints. You can specify it as a string, list, or dictionary:
{
"api_endpoints": {
"path_match": r"^/api/.*",
"headers": {
"Cache-Control": ["public", "max-age=3600"] # Cache for 1 hour
},
"order": 1
}
}By default the middleware will not set any Cache-Control header, so you should explicitly configure it to enable caching.
The package also provides authentication utilities for GitHub-based authentication and API key validation. See the auth.py module for detailed configuration options.
To enable Prometheus metrics in your FastAPI application, you can use the prometheus_fastapi_instrumentator package. Here's how to set it up:
from c2casgiutils import config
from prometheus_client import start_http_server
from prometheus_fastapi_instrumentator import Instrumentator
# Get Prometheus HTTP server port from environment variable 9000 by default
start_http_server(config.settings.prometheus.port)
instrumentator = Instrumentator(should_instrument_requests_inprogress=True)
instrumentator.instrument(app)To enable error tracking with Sentry in your application:
import os
import sentry_sdk
# Initialize Sentry if the URL is provided
if config.settings.sentry.dsn or 'SENTRY_DSN' in os.environ:
_LOGGER.info("Sentry is enabled with URL: %s", config.settings.sentry.url or os.environ.get("SENTRY_DSN"))
sentry_sdk.init(**{k: v for k, v in config.settings.sentry.model_dump().items() if v is not None and k != "tags"})
for tag, value in config.settings.sentry.tags.items():
sentry_sdk.set_tag(tag, value)Sentry will automatically capture exceptions and errors in your FastAPI application. For more advanced usage, refer to the Sentry Python SDK documentation and FastAPI integration guide.
The package includes some helpers to initialize applications from the command line. See the cli.py module for more details.
from c2casgiutils import cli
import asyncio
from argparse import ArgumentParser
async def main_() -> None:
"""Main entry point for CLI."""
parser = ArgumentParser(description="My Application CLI")
cli.add_arguments(parser)
args = parser.parse_args()
await cli.init(args)
# Initialize Sentry if the URL is provided
if config.settings.sentry.dsn or "SENTRY_DSN" in os.environ:
_LOGGER.info("Sentry is enabled with URL: %s", config.settings.sentry.dsn or os.environ.get("SENTRY_DSN"))
sentry_sdk.init(**config.settings.sentry.model_dump())
if c2casgiutils.config.settings.prometheus.port is not None:
prometheus_client.start_http_server(c2casgiutils.config.settings.prometheus.port)
# This method is required for console_scripts entry point
def main() -> None:
"""Main entry point for CLI."""
asyncio.run(main_())
if __name__ == "__main__":
main()See: https://github.com/camptocamp/c2casgiutils/blob/master/c2casgiutils/config.py
Optional, default value: None
Redis connection URL
Optional, default value: None
Redis connection options, e.g. 'socket_timeout=5,ssl=True'.
Optional, default value: None
Redis Sentinels
Optional, default value: None
Redis service name for Sentinel
Optional, default value: 0
Redis database number
Optional, default value: broadcast_api_
Redis prefix for broadcast channels
Optional, default value: c2casgiutils_
Prefix for Prometheus metrics
Optional, default value: 9000
Port for Prometheus metrics
Optional, default value: None
Sentry DSN
Optional, default value: False
Enable Sentry debug mode
Optional, default value: None
Sentry release version
Optional, default value: production
Sentry environment
Optional, default value: None
Sentry distribution
Optional, default value: 1.0
Sample rate for error events
Optional, default value: []
List of exception class names to ignore
Optional, default value: 100
Maximum number of breadcrumbs to capture
Optional, default value: False
Attach stack trace to all messages
Optional, default value: None
Send default PII
Optional, default value: None
Event scrubber for sensitive information
Optional, default value: True
Include source context in events
Optional, default value: True
Include local variables in events
Optional, default value: False
Add full stack trace to events
Optional, default value: 100
Maximum number of stack frames to capture
Optional, default value: None
Server name for Sentry events
Optional, default value: <working_directory>
Root directory of the project
Optional, default value: []
List of module prefixes that are in the app
Optional, default value: []
List of module prefixes that are not in the app
Optional, default value: medium
Maximum request body size to capture
Optional, default value: 1024
Maximum length of values in event payloads
Optional, default value: None
Path to alternative CA bundle file in PEM format
Optional, default value: True
Send client reports to Sentry
Optional, default value: {}
Default tags for Sentry events, loaded from environment variables with prefix C2C__SENTRY__TAG_ to set tags. The tag name will be the part after the prefix, converted to lowercase. For example, C2C__SENTRY__TAG_SERVICE=my-service will set a tag named service (lowercase) with value my-service.
Optional, default value: None
JWT secret key
Optional, default value: HS256
JWT algorithm (default: HS256)
Optional, default value: c2c-jwt-auth
Authentication cookie name
Optional, default value: 604800
Authentication cookie age in seconds (default: 7 days)
Optional, default value: lax
SameSite attribute for the JWT cookies (state and auth token). Defaults to 'lax' to support OAuth and other redirect-based login flows that rely on the cookie being sent on top-level navigation from external sites. Use 'strict' for stronger CSRF protection when such flows are not required.
lax, strict, none
Optional, default value: True
Whether the JWT cookie should be secure
Optional, default value: None
Path for the JWT cookie (default: the c2c index path)
Optional, default value: None
Secret key for trivial authentication (not secure)
Optional, default value: None
GitHub repository for authentication
Optional, default value: pull
GitHub access type
Optional, default value: https://github.com/login/oauth/authorize
GitHub OAuth authorization URL
Optional, default value: https://github.com/login/oauth/access_token
GitHub OAuth token URL
Optional, default value: https://api.github.com/user
GitHub user API URL
Optional, default value: https://api.github.com/repos
GitHub repository API URL
Optional, default value: None
GitHub OAuth client ID
Optional, default value: None
GitHub OAuth client secret
Optional, default value: repo
GitHub OAuth scope
Optional, default value: None
GitHub proxy URL
Optional, default value: c2c-state
GitHub state cookie name
Optional, default value: 600
GitHub state cookie age in seconds (default: 10 minutes)
Optional, default value: None
Test username
Optional, default value: c2c_logging_level_
Redis prefix for logging settings
Optional, default value: c2casgiutils
Application module name for logging
Optional, default value: none
Proxy headers mode: 'none' disables host/proto rewriting, 'x-forwarded' trusts X-Forwarded-* headers, 'forwarded' trusts RFC7239 Forwarded header
none, x-forwarded, forwarded
Optional, default value: ['127.0.0.1']
Trusted proxy client hosts/networks. Accepts comma-separated string or list (e.g. '127.0.0.1,10.0.0.0/8' or '*').
Optional, default value: False
The application is running in HTTP mode to be used for development only (default: False)
Optional, default value: /
Route prefix for the application, should start and end with a '/'