Connexion

repository·main·Indexed 26 days ago

https://github.com/spec-first/connexion

A modern Python web framework for spec-first and API-first development. Connexion uses OpenAPI (Swagger) specifications to automate route registration, request/response validation, and parameter injection. It supports both asynchronous applications via AsyncApp and synchronous Flask-based applications via FlaskApp, providing a CLI for testing and mocking OpenAPI specifications.

Tokens
27.6K
Snippets
69
Records
182
Agent score
88%

What's inside connexion

  1. Overview of Connexion features

    main

    Connexion is a spec-first Python web framework that uses an OpenAPI (Swagger) specification to drive API functionality. Based on your specification, it automatically provides:

    • Automatic routing: Maps API endpoints to your Python functions.
    • Authentication: Handles security requirements defined in the spec.
    • Request validation: Ensures incoming requests match the specification.
    • Parameter parsing and injection: Automatically extracts and passes parameters to your functions.
    • Response serialization: Converts Python objects to the specified response format.
    • Response validation: Validates that your application's responses adhere to the specification.
    • Swagger UI: Provides a live documentation console with a 'try it out' feature.

    Connexion also includes a CLI to test and mock your specification.

  2. Configure API Key Authentication

    main

    To implement API key authentication, include x-apikeyInfoFunc in your API security definition or set the APIKEYINFO_FUNC environment variable.

    The validation function must accept the following arguments:

    • apikey
    • required_scopes (optional)
    • request (optional)
  3. Run your Connexion application

    main

    If your application is defined as app in run.py, use uvicorn:

    $ uvicorn run:app

    For production, gunicorn with the uvicorn worker is recommended:

    $ gunicorn -k uvicorn.workers.UvicornWorker run:app

    Using app.run() (Development only)

    If you installed connexion[uvicorn], you can run the app directly in your code. To enable automatic reloading, pass the application as an import string:

    from pathlib import Path
    
    # Standard run
    app.run()
    
    # Run with automatic reloading
    app.run(f"{Path(__file__).stem}:app")
    uvicorn run:app
  4. Implement a security validation function

    main

    Connexion uses pluggable security validation functions to verify credentials and return user information.

    To register a validation function, you can either:

    1. Define it in your API security definition using the key x-{type}InfoFunc (e.g., x-basicInfoFunc).
    2. Set an environment variable named {TYPE}INFO_FUNC (e.g., BASICINFO_FUNC).

    Important: The function must be referenced as a string using the same syntax as operationId. If you use a resolver for operation IDs, you must specify the complete path to the security module.

    All validation functions must return a dictionary complying with RFC 7662. The returned information is made available to your endpoint view functions via the context or as a Context argument.

    {
      "active": true,
      "client_id": "l238j323ds-23ij4",
      "username": "jdoe",
      "scope": "read write dolphin",
      "sub": "Z5O3upPC88QrAjx00dis",
      "aud": "https://protected.example.net/resource",
      "iss": "https://server.example.com/",
      "exp": 1419356238,
      "iat": 1419350238,
      "extension_field": "twenty-seven"
    }
  5. Enable response serialization with ConnexionMiddleware

    main

    When using ConnexionMiddleware to wrap a third-party application, you must use specific decorators to enable automatic parameter injection and response serialization.

    • Use FlaskDecorator for Flask applications.
    • Use StarletteDecorator for Starlette applications.

    Note that WSGIDecorator and ASGIDecorator do not support response serialization by default.

    # For Flask
    from connexion import ConnexionMiddleware
    from connexion.decorators import FlaskDecorator
    from flask import Flask
    
    app = Flask(__name__)
    app = ConnexionMiddleware(app)
    app.add_api("openapi.yaml")
    
    @app.route("/endpoint")
    @FlaskDecorator()
    def endpoint(name):
        ...
    # For Starlette
    from connexion import ConnexionMiddleware
    from connexion.decorators import StarletteDecorator
    from starlette.applications import Starlette
    from starlette.routing import Route
    
    @StarletteDecorator()
    def endpoint(name):
        ...
    
    app = Starlette(routes=[Route('/endpoint', endpoint)])
    app = ConnexionMiddleware(app)
    app.add_api("openapi.yaml")
  6. Customize the default middleware stack

    main

    By default, Connexion includes a standard stack of middlewares (such as SecurityMiddleware, RequestValidationMiddleware, etc.). If you need to modify or remove specific default middlewares, you can pass a custom list to the middlewares argument during application instantiation.

    Example: Removing SecurityMiddleware to handle security via an external API Gateway.

    from connexion import AsyncApp, ConnexionMiddleware
    from connexion.middleware.security import SecurityMiddleware
    
    # Filter out SecurityMiddleware from the default list
    middlewares = [middleware for middleware in ConnexionMiddleware.default_middlewares
                   if middleware is not SecurityMiddleware]
    
    app = AsyncApp(__name__, middlewares=middlewares)
  7. Configure API versioning and basePath

    main

    To implement versioned APIs (e.g., http://{HOST}/1.0/hello_world), you can define a base path using one of the following methods depending on your specification version or application setup.

    OpenAPI 3

    Set the base URL path in the servers block of your specification. You can use a full URL or a relative path.

    Swagger 2.0

    Define a basePath at the top level of your Swagger 2.0 specification.

    Programmatic Base Path

    If you prefer not to include the base path in your specification file, you can provide it when adding the API to your application using the base_path argument in add_api.

    # OpenAPI 3 example
    servers:
      - url: https://{{HOST}}/1.0
        description: full url example
      - url: /1.0
        description: relative path example
    # Swagger 2.0 example
    basePath: /1.0
    # Programmatic approach
    app.add_api('openapi.yaml', base_path='/1.0')
  8. Register lifespan handlers in Connexion

    main

    You can register lifespan handlers to execute code before the application starts and after it shuts down. This is useful for managing resources like database connections or machine learning models.

    To implement a lifespan handler, create an asynchronous context manager using @contextlib.asynccontextmanager. The handler receives the application instance (e.g., AsyncApp, FlaskApp, or ConnexionMiddleware) and can yield a dictionary of state. This state becomes accessible on the request.state object during request processing.

    import contextlib
    import typing
    from connexion import AsyncApp, request
    
    @contextlib.asynccontextmanager
    async def lifespan_handler(app: AsyncApp) -> typing.AsyncIterator:
        # Startup logic
        client = Client()
        yield {"client": client}
        # Shutdown logic
        client.close()
    
    def route():
        # Access yielded state via request.state
        client = request.state.client
        client.call()
    
    app = AsyncApp(__name__, lifespan=lifespan_handler)
  9. Enable CORS in Connexion

    main

    You can enable Cross-Origin Resource Sharing (CORS) by adding the CORSMiddleware from Starlette to your Connexion application. It is recommended to add the middleware before the RoutingMiddleware. Use the add_middleware method and specify a position using MiddlewarePosition.

    ### For AsyncApp
    ```python
    from pathlib import Path
    
    from connexion import AsyncApp
    from connexion.middleware import MiddlewarePosition
    from starlette.middleware.cors import CORSMiddleware
    
    
    app = AsyncApp(__name__)
    
    app.add_middleware(
        CORSMiddleware,
        position=MiddlewarePosition.BEFORE_EXCEPTION,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    app.add_api("openapi.yaml")
    
    if __name__ == "__main__":
        app.run(f"{Path(__file__).stem}:app", port=8080)

    For FlaskApp

    from pathlib import Path
    
    from connexion import FlaskApp
    from connexion.middleware import MiddlewarePosition
    from starlette.middleware.cors import CORSMiddleware
    
    
    app = FlaskApp(__name__)
    
    app.add_middleware(
        CORSMiddleware,
        position=MiddlewarePosition.BEFORE_EXCEPTION,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    app.add_api("openapi.yaml")
    
    if __name__ == "__main__":
        app.run(f"{Path(__file__).stem}:app", port=8080)

    For ConnexionMiddleware (ASGI Frameworks)

    from pathlib import Path
    from asgi_framework import App
    from connexion import ConnexionMiddleware
    from starlette.middleware.cors import CORSMiddleware
    
    
    app = App(__name__)
    app = ConnexionMiddleware(app)
    
    app.add_middleware(
        CORSMiddleware,
        position=MiddlewarePosition.BEFORE_EXCEPTION,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )
    
    app.add_api("openapi.yaml")
    
    if __name__ == "__main__":
        app.run(f"{Path(__file__).stem}:app", port=8080)
  10. Register an API using an OpenAPI specification

    main

    Connexion automatically links OpenAPI (Swagger) specifications to Python view functions using the operationId. It handles routing, security, request/parameter parsing, and response serialization based on the spec.

    Use the add_api() method to register an API. If configuration options are provided to both the App and the API, the API-level values take precedence.

    Example Implementation:

    run.py

    def post_greeting(name: str):
        return f"Hello {name}", 200
    
    app.add_api("openapi.yaml")

    openapi.yaml

    openapi: "3.0.0"
    info:
      title: Greeting application
      version: 0.0.1
    paths:
      /greeting/{name}:
        post:
          operationId: run.post_greeting
          responses:
            '200':
              description: "Greeting response"
              content:
                text/plain:
                  schema:
                    type: string
          parameters:
            - name: name
              in: path
              required: true
              schema:
                type: string
    app.add_api("openapi.yaml")
  11. Run an OpenAPI specification with the Connexion CLI

    main

    Use the connexion run subcommand to start a server based on an OpenAPI specification. This is useful for verifying and inspecting your API design before implementing the actual operation handlers. You can provide a local YAML file or a URL to a specification file.

    $ connexion run your_api.yaml
  12. Enable strict parameter validation

    main

    By default, Connexion validates parameters (schema, type, format, range, required, nullability) against your OpenAPI specification. To disallow any extra parameters that are not explicitly defined in your specification, enable strict_validation. This can be set at both the application and the API level.

    If validation fails, Connexion returns a 400 Bad Request response.

    # For AsyncApp
    from connexion import AsyncApp
    app = AsyncApp(__name__, strict_validation=True)
    app.add_api("openapi.yaml", strict_validation=True)
    
    # For FlaskApp
    from connexion import FlaskApp
    app = FlaskApp(__name__, strict_validation=True)
    app.add_api("openapi.yaml", strict_validation=True)
    
    # For ConnexionMiddleware
    from asgi_framework import App
    from connexion import ConnexionMiddleware
    app = App(__name__)
    app = ConnexionMiddleware(app, strict_validation=True)
    app.add_api("openapi.yaml", strict_validation=True)