aetos/openapi-first
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

standardize packaging, tooling, docs, CI, and licensing

Browse Source
This commit is contained in:
Vishesh 'ironeagle' Bangotra
2026-09-10 18:49:04 +05:30
parent c95de5a6e9
commit 0efe68b052
25 changed files with 667 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
.drone.yml
View File

@@ -32,6 +32,26 @@ steps:
echo "🆕 New version detected: $PACKAGE_NAME==$VERSION"
fi
- name: quality-gate
image: python:3.13-slim
environment:
PIP_REPO_URL:
from_secret: PIP_REPO_URL
PIP_USERNAME:
from_secret: PIP_USERNAME
PIP_PASSWORD:
from_secret: PIP_PASSWORD
commands:
- pip install --upgrade pip build
- |
AUTH_URL="https://${PIP_USERNAME}:${PIP_PASSWORD}@$(echo "${PIP_REPO_URL#*://}" | sed 's:/*$::')/simple"
pip install --index-url "$AUTH_URL" --extra-index-url https://pypi.org/simple/ -U ".[dev]"
- echo "🛡️ Running quality gate..."
- python -m black --check .
- python -m ruff check .
- python -m mypy
- python -m pytest
- name: build-package
image: python:3.13-slim
commands:
@@ -126,4 +146,4 @@ steps:
trigger:
event:
- custom
- custom

21
CHANGELOG.md Normal file
View File

@@ -0,0 +1,21 @@
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [Unreleased]
### Added
- `py.typed` marker for PEP 561 type information.
- `.drone.yml` CI with a quality-gate step (black, ruff, mypy, pytest).
- Canonical `docforge.nav.yml`, generated `mkdocs.yml` and `mcp_docs/` via doc-forge.
- MIT `LICENSE`.
### Changed
- Standardized `pyproject.toml` (canonical packaging, lint tool config, extras).
- `templates/` excluded from black/ruff scans.
### Fixed
- Stub fixes for typed API surfaces.

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 Aetoskia Platform
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

4
openapi_first/__init__.pyi
View File

@@ -1,9 +1,9 @@
from . import app as app
from . import binder as binder
from . import loader as loader
from . import client as client
from . import errors as errors
from . import codegen as codegen
from . import codegen_routes as codegen_routes
from . import errors as errors
from . import loader as loader
__all__ = ["app", "binder", "loader", "client", "errors", "codegen", "codegen_routes"]

2
openapi_first/app.py
View File

@@ -46,8 +46,10 @@ _env_pattern = re.compile(r"\{(\w+)\}")
def _resolve_env(value: str) -> str:
"""Replace {ENV_VAR} placeholders with values from os.environ."""
def _replace(m: re.Match) -> str:
return os.environ.get(m.group(1), m.group(0))
return _env_pattern.sub(_replace, value)

7
openapi_first/app.pyi
View File

@@ -1,5 +1,8 @@
from fastapi import FastAPI
from typing import Any
from fastapi import FastAPI
class OpenAPIFirstApp(FastAPI):
def __init__(self, *, openapi_path: str, routes_module: Any, **fastapi_kwargs: Any) -> None: ...
def __init__(
self, *, openapi_path: str, routes_module: Any, **fastapi_kwargs: Any
) -> None: ...

8
openapi_first/binder.py
View File

@@ -85,8 +85,14 @@ def bind_routes(
paths = spec.get("paths", {})
security_deps = security_deps or {}
http_methods = {"get", "put", "post", "delete", "options", "head", "patch", "trace"}
for path, methods in paths.items():
for http_method, operation in methods.items():
if http_method.lower() not in http_methods or not isinstance(
operation, dict
):
continue
operation_id = operation.get("operationId")
if not operation_id:
@@ -102,7 +108,7 @@ def bind_routes(
path=path,
method=http_method,
operation_id=operation_id,
)
) from None
key = f"{http_method.upper()}:{path}"
deps = security_deps.get(key, [])

5
openapi_first/binder.pyi
View File

@@ -1,4 +1,5 @@
from typing import Any, Dict
from typing import Any
from fastapi import FastAPI
def bind_routes(app: FastAPI, spec: Dict[str, Any], routes_module: Any) -> None: ...
def bind_routes(app: FastAPI, spec: dict[str, Any], routes_module: Any) -> None: ...

14
openapi_first/cli.py
View File

@@ -11,9 +11,8 @@ bundled templates packaged with the library.
import argparse
import shutil
from pathlib import Path
from importlib import resources
from pathlib import Path
DEFAULT_TEMPLATE = "health_app"
@@ -52,14 +51,15 @@ def copy_template(template: str, target_dir: Path) -> None:
target_dir.mkdir(parents=True, exist_ok=True)
root = resources.files("openapi_first.templates")
src = root / template
if not src.exists():
if template not in available_templates():
raise FileNotFoundError(
f"Template '{template}' not found. "
f"Available templates: {', '.join(available_templates())}"
)
src = root / template
with resources.as_file(src) as path:
shutil.copytree(path, target_dir, dirs_exist_ok=True)
@@ -168,11 +168,11 @@ def main() -> None:
# Handle the case where someone uses the old CLI style (openapi-first template path)
# argparse with subparsers might not automatically handle this if positional args are passed
# but let's assume we want to encourage the new style.
# If no command was provided but positional args were, they might be for scaffolding
# This is a bit tricky with argparse subparsers.
# This is a bit tricky with argparse subparsers.
# For simplicity, let's just support the new explicit commands.
if args.command is None and not any(vars(args).values()):
parser.print_help()
return

33
openapi_first/client.py
View File

@@ -31,7 +31,8 @@ Notes:
- This module intentionally does NOT: Generate client code, validate request/response schemas, deserialize responses, retry requests, implement authentication helpers, or assume non-2xx responses are failures.
"""
from typing import Any, Callable, Dict, Optional
from collections.abc import Callable
from typing import Any
from urllib.parse import urljoin
import httpx
@@ -88,8 +89,8 @@ class OpenAPIClient:
def __init__(
self,
spec: dict[str, Any],
base_url: Optional[str] = None,
client: Optional[httpx.Client] = None,
base_url: str | None = None,
client: httpx.Client | None = None,
) -> None:
"""
Initialize the OpenAPI client.
@@ -110,7 +111,7 @@ class OpenAPIClient:
self.base_url = base_url or self._resolve_base_url(spec)
self.client = client or httpx.Client(base_url=self.base_url)
self._operations: Dict[str, Callable[..., httpx.Response]] = {}
self._operations: dict[str, Callable[..., httpx.Response]] = {}
self._build_operations()
# ------------------------------------------------------------------ #
@@ -123,7 +124,7 @@ class OpenAPIClient:
except KeyError:
raise AttributeError(f"No such operationId: {name}") from None
def operations(self) -> Dict[str, Callable[..., httpx.Response]]:
def operations(self) -> dict[str, Callable[..., httpx.Response]]:
return dict(self._operations)
# ------------------------------------------------------------------ #
@@ -149,7 +150,13 @@ class OpenAPIClient:
for path, path_item in paths.items():
for method, operation in path_item.items():
if method.lower() not in {
"get", "post", "put", "patch", "delete", "head", "options"
"get",
"post",
"put",
"patch",
"delete",
"head",
"options",
}:
continue
@@ -181,11 +188,11 @@ class OpenAPIClient:
def call(
*,
path_params: Optional[dict[str, Any]] = None,
query: Optional[dict[str, Any]] = None,
headers: Optional[dict[str, str]] = None,
body: Optional[Any] = None,
timeout: Optional[float] = None,
path_params: dict[str, Any] | None = None,
query: dict[str, Any] | None = None,
headers: dict[str, str] | None = None,
body: Any | None = None,
timeout: float | None = None,
) -> httpx.Response:
url = self._build_url(path, path_params or {})
@@ -228,9 +235,7 @@ class OpenAPIClient:
try:
formatted_path = path.format(**path_params)
except KeyError as exc:
raise OpenAPIClientError(
f"Missing path parameter: {exc.args[0]}"
) from exc
raise OpenAPIClientError(f"Missing path parameter: {exc.args[0]}") from exc
return urljoin(self.base_url, formatted_path.lstrip("/"))

16
openapi_first/client.pyi
View File

@@ -1,13 +1,21 @@
from typing import Any, Callable, Dict, Optional
from collections.abc import Callable
from typing import Any
import httpx
from .errors import OpenAPIFirstError
class OpenAPIClientError(OpenAPIFirstError): ...
class OpenAPIClient:
spec: Dict[str, Any]
spec: dict[str, Any]
base_url: str
client: httpx.Client
def __init__(self, spec: Dict[str, Any], base_url: Optional[str] = ..., client: Optional[httpx.Client] = ...) -> None: ...
def __init__(
self,
spec: dict[str, Any],
base_url: str | None = ...,
client: httpx.Client | None = ...,
) -> None: ...
def __getattr__(self, name: str) -> Callable[..., httpx.Response]: ...
def operations(self) -> Dict[str, Callable[..., httpx.Response]]: ...
def operations(self) -> dict[str, Callable[..., httpx.Response]]: ...

1
openapi_first/codegen.py
View File

@@ -9,6 +9,7 @@ definitions.
"""
from pathlib import Path
from datamodel_code_generator import (
InputFileType,
PythonVersion,

43
openapi_first/codegen_routes.py
View File

@@ -66,7 +66,7 @@ def generate_routes(
output_dir.mkdir(parents=True, exist_ok=True)
# Group paths by resource (first non-param path segment)
resources: dict[str, list[tuple[str, str, dict]]] = {}
resources_map: dict[str, list[tuple[str, str, dict[str, Any]]]] = {}
paths = spec.get("paths", {})
for path, methods in paths.items():
@@ -74,18 +74,18 @@ def generate_routes(
if not segments:
continue
resource = segments[0]
if resource not in resources:
resources[resource] = []
if resource not in resources_map:
resources_map[resource] = []
for http_method, operation in methods.items():
if http_method.startswith("x-"):
continue
resources[resource].append((path, http_method, operation))
resources_map[resource].append((path, http_method, operation))
generated_files: list[Path] = []
for resource in sorted(resources):
operations = resources[resource]
for resource in sorted(resources_map):
operations = resources_map[resource]
_validate_operations(resource, operations)
file_path = output_dir / f"{resource}.py"
@@ -116,7 +116,9 @@ _TYPE_MAP: dict[str, str] = {
}
def _validate_operations(resource: str, operations: list[tuple[str, str, dict]]) -> None:
def _validate_operations(
resource: str, operations: list[tuple[str, str, dict[str, Any]]]
) -> None:
"""Ensure every operation has an operationId."""
for path, http_method, operation in operations:
if not operation.get("operationId"):
@@ -144,7 +146,7 @@ def _get_request_body_schema(operation: dict[str, Any]) -> str | None:
schema = media_info.get("schema", {})
ref = schema.get("$ref", "")
if ref:
return ref.rsplit("/", 1)[-1]
return str(ref.rsplit("/", 1)[-1])
return None
@@ -159,7 +161,7 @@ def _get_success_status(operation: dict[str, Any]) -> str:
return "200"
def _needs_any(operations: list[tuple[str, str, dict]]) -> bool:
def _needs_any(operations: list[tuple[str, str, dict[str, Any]]]) -> bool:
"""Check if any operation uses a type that requires `from typing import Any`."""
for _, _, op in operations:
for param in op.get("parameters", []):
@@ -176,7 +178,7 @@ def _needs_any(operations: list[tuple[str, str, dict]]) -> bool:
def _generate_resource_file(
resource: str,
operations: list[tuple[str, str, dict]],
operations: list[tuple[str, str, dict[str, Any]]],
spec_path: str,
use_models: bool,
models_module: str,
@@ -205,7 +207,9 @@ def _generate_resource_file(
if schema:
schemas_needed.add(schema)
if schemas_needed:
lines.append(f"from {models_module} import {', '.join(sorted(schemas_needed))}")
lines.append(
f"from {models_module} import {', '.join(sorted(schemas_needed))}"
)
lines.append("")
@@ -237,7 +241,6 @@ def _generate_handler(
schema: dict[str, Any] = param.get("schema", {})
param_type: str = _resolve_type(schema)
required: bool = param.get("required", False)
description: str = schema.get("description", schema.get("x-description", ""))
default_raw = schema.get("default")
if param_in == "path":
@@ -277,13 +280,15 @@ def _generate_handler(
# Document parameters
doc_params: list[tuple[str, str, str]] = []
for param in operation.get("parameters", []):
name: str = param.get("name", "")
param_in: str = param.get("in", "")
schema: dict[str, Any] = param.get("schema", {})
param_type: str = _resolve_type(schema)
description: str = param.get("description", schema.get("x-description", ""))
if param_in in ("path", "query"):
doc_params.append((name, param_type, description))
doc_name: str = param.get("name", "")
doc_in: str = param.get("in", "")
doc_schema: dict[str, Any] = param.get("schema", {})
doc_type: str = _resolve_type(doc_schema)
doc_description: str = param.get(
"description", doc_schema.get("x-description", "")
)
if doc_in in ("path", "query"):
doc_params.append((doc_name, doc_type, doc_description))
if doc_params:
lines.append("")

7
openapi_first/errors.py
View File

@@ -18,6 +18,7 @@ Notes:
These errors should normally cause immediate application failure.
"""
class OpenAPIFirstError(Exception):
"""
Base exception for all OpenAPI-first enforcement errors.
@@ -31,6 +32,7 @@ class OpenAPIFirstError(Exception):
- All exceptions raised by the OpenAPI-first core should inherit
from this type.
"""
pass
@@ -70,9 +72,6 @@ class MissingOperationHandler(OpenAPIFirstError):
f"({method.upper()} {path})"
)
else:
message = (
f"Missing operationId for operation "
f"({method.upper()} {path})"
)
message = f"Missing operationId for operation " f"({method.upper()} {path})"
super().__init__(message)

6
openapi_first/errors.pyi
View File

@@ -1,6 +1,6 @@
from typing import Optional
class OpenAPIFirstError(Exception): ...
class MissingOperationHandler(OpenAPIFirstError):
def __init__(self, *, path: str, method: str, operation_id: Optional[str] = ...) -> None: ...
def __init__(
self, *, path: str, method: str, operation_id: str | None = ...
) -> None: ...

1
openapi_first/loader.py
View File

@@ -46,6 +46,7 @@ class OpenAPISpecLoadError(OpenAPIFirstError):
- This error indicates that the OpenAPI document is unreadable,
malformed, or violates the OpenAPI 3.x specification.
"""
pass

4
openapi_first/loader.pyi
View File

@@ -1,8 +1,8 @@
from pathlib import Path
from typing import Any, Dict, Union
from typing import Any
from .errors import OpenAPIFirstError
class OpenAPISpecLoadError(OpenAPIFirstError): ...
def load_openapi(path: Union[str, Path]) -> Dict[str, Any]: ...
def load_openapi(path: str | Path) -> dict[str, Any]: ...

0
openapi_first/py.typed Normal file
View File

27
openapi_first/security.py
View File

@@ -8,10 +8,11 @@ FastAPI dependencies for token validation (e.g., Bearer JWT introspection).
import os
import re
from typing import Any, Callable
from collections.abc import Callable
from typing import Any
import httpx
from fastapi import Depends, Request, HTTPException
from fastapi import Depends, HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
_env_pattern = re.compile(r"\{(\w+)\}")
@@ -19,14 +20,16 @@ _env_pattern = re.compile(r"\{(\w+)\}")
def _resolve_env(value: str) -> str:
"""Replace {ENV_VAR} placeholders with values from os.environ."""
def _replace(m: re.Match) -> str:
def _replace(m: re.Match[str]) -> str:
return os.environ.get(m.group(1), "")
return _env_pattern.sub(_replace, value)
def _resolve_scheme(scheme: dict) -> dict:
def _resolve_scheme(scheme: dict[str, Any]) -> dict[str, Any]:
"""Recursively resolve env vars in all string-valued fields of a scheme."""
resolved = {}
resolved: dict[str, Any] = {}
for key, value in scheme.items():
if isinstance(value, str):
resolved[key] = _resolve_env(value)
@@ -37,13 +40,13 @@ def _resolve_scheme(scheme: dict) -> dict:
return resolved
def parse_security_schemes(spec: dict) -> dict[str, dict]:
def parse_security_schemes(spec: dict[str, Any]) -> dict[str, dict[str, Any]]:
"""Extract and resolve environment variables in security schemes."""
raw = spec.get("components", {}).get("securitySchemes", {})
return {name: _resolve_scheme(scheme) for name, scheme in raw.items()}
def _make_bearer_dependency(introspect_url: str | None) -> Callable:
def _make_bearer_dependency(introspect_url: str | None) -> Callable[..., Any]:
"""
Create a FastAPI dependency that validates a Bearer JWT.
@@ -76,7 +79,7 @@ def _make_bearer_dependency(introspect_url: str | None) -> Callable:
raise HTTPException(
status_code=503,
detail="Authentication service unavailable",
)
) from None
if resp.status_code != 200:
raise HTTPException(status_code=401, detail="Invalid or expired token")
@@ -92,7 +95,9 @@ def _make_bearer_dependency(introspect_url: str | None) -> Callable:
return _bearer_dep
def _build_dependency(scheme_name: str, scheme: dict) -> Callable | None:
def _build_dependency(
scheme_name: str, scheme: dict[str, Any]
) -> Callable[..., Any] | None:
"""Return a FastAPI dependency callable for *scheme*, or ``None``."""
scheme_type = scheme.get("type")
@@ -106,8 +111,8 @@ def _build_dependency(scheme_name: str, scheme: dict) -> Callable | None:
def make_security_dependencies(
spec: dict,
security_schemes: dict[str, dict],
spec: dict[str, Any],
security_schemes: dict[str, dict[str, Any]],
) -> dict[str, list[Any]]:
"""
Build a mapping of ``METHOD:/path`` → list of ``Depends(...)``.

102
pyproject.toml
View File

@@ -37,9 +37,11 @@ classifiers = [
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Framework :: FastAPI",
"Topic :: Software Development :: Libraries",
"Topic :: Internet :: WWW/HTTP :: HTTP Servers",
"Typing :: Typed",
]
@@ -66,9 +68,15 @@ openapi-first = "openapi_first.cli:main"
[project.optional-dependencies]
dev = [
"pytest>=8.0.0",
"pytest-asyncio>=0.21.0",
"pytest-cov>=4.1.0",
"black>=23.0.0",
"ruff>=0.3.0",
"mypy>=1.8.0",
"build>=1.0.0",
"twine>=4.0.0",
"pre-commit>=3.4.0",
"doc-forge[mcp,mkdocs]>=0.0.6",
]
docs = [
@@ -77,6 +85,10 @@ docs = [
"mkdocstrings[python]>=0.24.0",
]
all = [
"openapi-first[dev,docs]",
]
[project.urls]
Homepage = "https://git.aetoskia.com/aetos/openapi-first"
@@ -90,15 +102,101 @@ Versions = "https://git.aetoskia.com/aetos/openapi-first/tags"
packages = { find = { include = ["openapi_first*"] } }
[tool.setuptools.package-data]
openapi_first = ["templates/**/*"]
openapi_first = ["py.typed", "templates/**/*"]
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto"
python_files = ["test_*.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
addopts = [
"--strict-markers",
"--strict-config",
"--cov=openapi_first",
"--cov-report=term-missing",
"--cov-report=html",
"--cov-report=xml",
]
[tool.black]
line-length = 88
target-version = ["py310", "py311", "py312", "py313"]
include = '\.pyi?$'
extend-exclude = '''
/(
\.eggs
| \.git
| \.hg
| \.mypy_cache
| \.tox
| \.venv
| build
| dist
| openapi_first/templates
)/
'''
[tool.ruff]
line-length = 100
line-length = 88
target-version = "py310"
exclude = ["openapi_first/templates"]
[tool.ruff.lint]
select = [
"E",
"W",
"F",
"I",
"B",
"C4",
"UP",
]
ignore = [
"E501",
"B008",
"C901",
]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401", "I001"]
"tests/*" = ["B008"]
[tool.mypy]
python_version = "3.10"
strict = true
exclude = [
"openapi_first/templates/",
"tests/",
]
files = ["openapi_first"]
[[tool.mypy.overrides]]
module = [
"openapi_spec_validator.*",
"datamodel_code_generator.*",
]
ignore_missing_imports = true
[tool.coverage.run]
source = ["openapi_first"]
omit = [
"*/tests/*",
"*/test_*.py",
]
[tool.coverage.report]
exclude_lines = [
"pragma: no cover",
"def __repr__",
"if __name__ == .__main__.:",
"raise AssertionError",
"raise NotImplementedError",
"if TYPE_CHECKING:",
"@abstractmethod",
]

43
tests/conftest.py Normal file
View File

@@ -0,0 +1,43 @@
"""Shared fixtures for the openapi-first smoke test suite."""
import json
import pytest
MINIMAL_SPEC = {
"openapi": "3.0.3",
"info": {"title": "Smoke Test API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/v1"}],
"paths": {
"/health": {
"get": {
"operationId": "get_health",
"responses": {"200": {"description": "OK"}},
}
}
},
}
@pytest.fixture
def minimal_spec() -> dict:
"""Return a minimal valid OpenAPI 3.0.3 specification."""
return json.loads(json.dumps(MINIMAL_SPEC))
@pytest.fixture
def spec_file(tmp_path, minimal_spec):
"""Write the minimal spec to a temp JSON file and return its path."""
def _write(name: str, spec: dict | None = None) -> str:
path = tmp_path / name
payload = spec if spec is not None else minimal_spec
if name.endswith(".json"):
path.write_text(json.dumps(payload), encoding="utf-8")
else:
import yaml
path.write_text(yaml.safe_dump(payload), encoding="utf-8")
return str(path)
return _write

89
tests/test_app.py Normal file
View File

@@ -0,0 +1,89 @@
"""Smoke tests for openapi_first.app."""
import types
import pytest
from fastapi.testclient import TestClient
from openapi_first.app import OpenAPIFirstApp
from openapi_first.errors import MissingOperationHandler
SPEC = {
"openapi": "3.0.3",
"info": {"title": "App API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/v1"}],
"paths": {
"/health": {
"get": {
"operationId": "get_health",
"responses": {"200": {"description": "OK"}},
}
},
"/greet/{name}": {
"get": {
"operationId": "get_greet",
"parameters": [
{
"name": "name",
"in": "path",
"required": True,
"schema": {"type": "string"},
}
],
"responses": {"200": {"description": "OK"}},
}
},
},
}
def _routes_module():
module = types.ModuleType("smoke_routes")
def get_health():
return {"status": "ok"}
def get_greet(name: str):
return {"greeting": f"Hello, {name}"}
module.get_health = get_health
module.get_greet = get_greet
return module
@pytest.fixture
def spec_file(tmp_path):
import json
path = tmp_path / "openapi.json"
path.write_text(json.dumps(SPEC), encoding="utf-8")
return str(path)
def test_app_routes_served(spec_file):
app = OpenAPIFirstApp(openapi_path=spec_file, routes_module=_routes_module())
client = TestClient(app)
response = client.get("/health")
assert response.status_code == 200
assert response.json() == {"status": "ok"}
def test_app_path_param_route(spec_file):
app = OpenAPIFirstApp(openapi_path=spec_file, routes_module=_routes_module())
client = TestClient(app)
response = client.get("/greet/Ada")
assert response.status_code == 200
assert response.json() == {"greeting": "Hello, Ada"}
def test_app_overrides_openapi_schema(spec_file):
app = OpenAPIFirstApp(openapi_path=spec_file, routes_module=_routes_module())
openapi = app.openapi()
assert openapi["info"]["title"] == "App API"
assert "/health" in openapi["paths"]
def test_app_missing_handler_fails_at_startup(spec_file):
empty = types.ModuleType("empty_routes")
with pytest.raises(MissingOperationHandler):
OpenAPIFirstApp(openapi_path=spec_file, routes_module=empty)

72
tests/test_binder.py Normal file
View File

@@ -0,0 +1,72 @@
"""Smoke tests for openapi_first.binder."""
import types
import pytest
from fastapi import FastAPI
from openapi_first.binder import bind_routes
from openapi_first.errors import MissingOperationHandler
def _routes_module():
module = types.ModuleType("smoke_routes")
def get_health():
return {"status": "ok"}
module.get_health = get_health
return module
def test_bind_routes_registers_matching_routes(minimal_spec):
from fastapi.routing import APIRoute
app = FastAPI()
bind_routes(app, minimal_spec, _routes_module())
bound = [r for r in app.router.routes if isinstance(r, APIRoute)]
assert len(bound) == 1
assert bound[0].name == "get_health"
assert bound[0].path == "/health"
assert "GET" in bound[0].methods
def test_bind_routes_missing_handler(minimal_spec):
app = FastAPI()
module = types.ModuleType("empty_routes")
with pytest.raises(MissingOperationHandler):
bind_routes(app, minimal_spec, module)
def test_bind_routes_missing_operation_id(spec_file):
spec = {
"openapi": "3.0.3",
"info": {"title": "No OpId", "version": "1.0.0"},
"paths": {"/health": {"get": {"responses": {"200": {"description": "OK"}}}}},
}
app = FastAPI()
with pytest.raises(MissingOperationHandler):
bind_routes(app, spec, _routes_module())
def test_bind_routes_ignores_spec_only_paths():
from fastapi.routing import APIRoute
spec = {
"openapi": "3.0.3",
"info": {"title": "Path Item", "version": "1.0.0"},
"paths": {
"/health": {
"parameters": [{"name": "x-trace", "in": "header"}],
"get": {
"operationId": "get_health",
"responses": {"200": {"description": "OK"}},
},
}
},
}
app = FastAPI()
bind_routes(app, spec, _routes_module())
bound = [r for r in app.router.routes if isinstance(r, APIRoute)]
assert len(bound) == 1
assert bound[0].name == "get_health"

130
tests/test_client.py Normal file
View File

@@ -0,0 +1,130 @@
"""Smoke tests for openapi_first.client."""
import types
import pytest
from openapi_first.client import OpenAPIClient, OpenAPIClientError
SPEC = {
"openapi": "3.0.3",
"info": {"title": "Client API", "version": "1.0.0"},
"servers": [{"url": "https://api.example.com/v1"}],
"paths": {
"/users/{user_id}": {
"get": {
"operationId": "get_user",
"parameters": [
{
"name": "user_id",
"in": "path",
"required": True,
"schema": {"type": "integer"},
}
],
"responses": {"200": {"description": "OK"}},
}
},
"/users": {
"post": {
"operationId": "create_user",
"requestBody": {
"content": {"application/json": {"schema": {"type": "object"}}}
},
"responses": {"201": {"description": "Created"}},
}
},
},
}
class _RecordingClient:
"""Minimal httpx.Client stand-in that records and routes requests."""
def __init__(self, base_url=None):
self.base_url = base_url
self.requests = []
def request(
self,
*,
method,
url,
params=None,
headers=None,
json=None,
content=None,
timeout=None,
):
self.requests.append(
{
"method": method,
"url": url,
"params": params,
"headers": headers,
"json": json,
"content": content,
"timeout": timeout,
}
)
response = types.SimpleNamespace(status_code=200, json=lambda: {"ok": True})
return response
def test_default_base_url_from_servers():
client = OpenAPIClient(SPEC)
assert client.base_url == "https://api.example.com/v1/"
def test_operation_methods_are_generated(minimal_spec):
client = OpenAPIClient(minimal_spec, base_url="http://test")
assert "get_health" in client.operations()
assert callable(client.get_health)
def test_unknown_operation_raises_attribute_error(minimal_spec):
client = OpenAPIClient(minimal_spec, base_url="http://test")
with pytest.raises(AttributeError):
client.does_not_exist()
def test_path_param_formatting_and_urljoin():
rec = _RecordingClient()
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
client.get_user(path_params={"user_id": 42})
assert rec.requests[0]["url"] == "https://host/users/42"
def test_missing_required_path_param():
rec = _RecordingClient()
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
with pytest.raises(OpenAPIClientError):
client.get_user(path_params={})
def test_json_body_defaults_content_type():
rec = _RecordingClient()
client = OpenAPIClient(SPEC, base_url="https://host", client=rec)
client.create_user(body={"name": "Ada"})
req = rec.requests[0]
assert req["json"] == {"name": "Ada"}
assert req["headers"].get("Content-Type") == "application/json"
def test_missing_server_raises():
spec = dict(SPEC)
spec.pop("servers")
with pytest.raises(OpenAPIClientError):
OpenAPIClient(spec)
def test_duplicate_operation_id_raises():
spec = dict(SPEC)
spec["paths"]["/extra"] = {
"get": {
"operationId": "get_user",
"responses": {"200": {"description": "OK"}},
}
}
with pytest.raises(OpenAPIClientError):
OpenAPIClient(spec)

63
tests/test_loader.py Normal file
View File

@@ -0,0 +1,63 @@
"""Smoke tests for openapi_first.loader."""
import pytest
from openapi_first.loader import OpenAPISpecLoadError, load_openapi
def test_load_openapi_json(spec_file, minimal_spec):
path = spec_file("openapi.json")
spec = load_openapi(path)
assert spec == minimal_spec
def test_load_openapi_yaml(spec_file, minimal_spec):
path = spec_file("openapi.yaml")
spec = load_openapi(path)
assert spec == minimal_spec
def test_load_missing_file(tmp_path):
with pytest.raises(OpenAPISpecLoadError):
load_openapi(tmp_path / "missing.json")
def test_load_unsupported_extension(tmp_path):
path = tmp_path / "openapi.txt"
path.write_text("{}", encoding="utf-8")
with pytest.raises(OpenAPISpecLoadError):
load_openapi(path)
def test_load_invalid_spec(tmp_path):
path = tmp_path / "invalid.yaml"
path.write_text("not: [valid: openapi", encoding="utf-8")
with pytest.raises(OpenAPISpecLoadError):
load_openapi(path)
def test_load_validation_failure(tmp_path):
path = tmp_path / "bad.yaml"
path.write_text("openapi: 3.0.3\ninfo: {}\n", encoding="utf-8")
with pytest.raises(OpenAPISpecLoadError):
load_openapi(path)
def test_env_var_resolution(spec_file, monkeypatch):
monkeypatch.setenv("AUTH_HOST", "http://auth.internal")
spec = {
"openapi": "3.0.3",
"info": {"title": "Env API", "version": "1.0.0"},
"servers": [{"url": "{AUTH_HOST}/v1"}],
"paths": {
"/ping": {
"get": {
"operationId": "get_ping",
"responses": {"200": {"description": "OK"}},
}
}
},
}
path = spec_file("env.yaml", spec)
loaded = load_openapi(path)
assert loaded["servers"][0]["url"] == "{AUTH_HOST}/v1"