Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 150 additions & 0 deletions api/api/openapi.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import re
from typing import TYPE_CHECKING, Any, Literal

from drf_spectacular import generators, openapi
Expand Down Expand Up @@ -213,3 +214,152 @@ def get_security_definition(
"<a href='https://docs.flagsmith.com/clients/rest#private-api-endpoints'>Find out more</a>."
),
}


# Tag definitions controlling the order and display of sections in the Swagger UI.
TAGS: list[dict[str, str]] = [
{
"name": "Authentication",
"description": "Authentication, MFA, OAuth, and token management.",
},
{
"name": "Organisations",
"description": "Manage organisations, users, groups, invites, and API keys.",
},
{
"name": "Projects",
"description": "Manage projects, tags, and imports/exports.",
},
{
"name": "Environments",
"description": "Manage environments, API keys, and metrics.",
},
{
"name": "Features",
"description": "Manage features and multivariate options.",
},
{
"name": "Feature states",
"description": "Manage feature states and feature versioning.",
},
{
"name": "Identities",
"description": "Manage identities and traits.",
},
{
"name": "Segments",
"description": "Manage segments and segment rules.",
},
{
"name": "Integrations",
"description": "Configure third-party integrations (Amplitude, DataDog, Slack, etc.).",
},
{
"name": "Permissions",
"description": "Manage user and group permissions across organisations, projects, and environments.",
},
{
"name": "Webhooks",
"description": "Manage webhooks for organisations and environments.",
},
{
"name": "Audit",
"description": "Access audit logs.",
},
{
"name": "Analytics",
"description": "SDK analytics and telemetry.",
},
{
"name": "Metadata",
"description": "Manage metadata fields and model configuration.",
},
{
"name": "Onboarding",
"description": "Onboarding flows.",
},
{
"name": "Admin dashboard",
"description": "Platform hub admin dashboard endpoints.",
},
{
"name": "sdk",
"description": "SDK endpoints for flags, identities, and traits.",
},
{
"name": "mcp",
"description": "MCP-compatible endpoints.",
},
{
"name": "experimental",
"description": "Experimental endpoints subject to change.",
},
{
"name": "Other",
"description": "Other endpoints.",
},
]

# Ordered list of (regex, tag) rules for assigning tags to API operations.
# The first matching rule wins, so more specific patterns must come before
# broader ones (e.g. /analytics/ before /flags/).
_TAG_RULES: list[tuple[re.Pattern[str], str]] = [
(re.compile(r"/integrations/"), "Integrations"),
(re.compile(r"/user-permissions/|/user-group-permissions/"), "Permissions"),
(re.compile(r"/identities/|/edge-identities|/traits/"), "Identities"),
(re.compile(r"/featurestates/|feature-version|/feature-health/"), "Feature states"),
(re.compile(r"/analytics/"), "Analytics"),
(re.compile(r"/features/|/multivariate/|/flags/"), "Features"),
(re.compile(r"/segments/"), "Segments"),
(re.compile(r"/metadata/"), "Metadata"),
(re.compile(r"/audit/"), "Audit"),
(re.compile(r"/webhooks?/|cb-webhook|github-webhook"), "Webhooks"),
(re.compile(r"/auth/|/users/"), "Authentication"),
(re.compile(r"/onboarding/"), "Onboarding"),
(re.compile(r"/admin/dashboard/"), "Admin dashboard"),
(re.compile(r"/environments/"), "Environments"),
(re.compile(r"/organisations/"), "Organisations"),
(re.compile(r"/projects/"), "Projects"),
]

_EXCLUDED_PATHS: set[str] = {
"/api/v1/swagger.json",
"/api/v1/swagger.yaml",
}


def preprocessing_filter_spec(
endpoints: list[tuple[str, str, str, Any]],
) -> list[tuple[str, str, str, Any]]:
"""Filter out internal endpoints that should not appear in the API docs."""
return [
(path, path_regex, method, callback)
for path, path_regex, method, callback in endpoints
if path not in _EXCLUDED_PATHS
]


def postprocessing_assign_tags(
result: dict[str, Any], generator: Any, **kwargs: Any
) -> dict[str, Any]:
"""Assign descriptive tags to operations based on URL path patterns.

Only reassigns the default 'api' tag; operations with explicit tags
(sdk, mcp, experimental, etc.) are left unchanged.
"""
for path, path_item in result.get("paths", {}).items():
for method, operation in path_item.items():
if not isinstance(operation, dict):
continue
tags = operation.get("tags", [])
if tags != ["api"]:
continue
for pattern, tag in _TAG_RULES:
if pattern.search(path):
operation["tags"] = [tag]
break
else:
operation["tags"] = ["Other"]

result["tags"] = TAGS
return result
7 changes: 7 additions & 0 deletions api/app/settings/common.py
Original file line number Diff line number Diff line change
Expand Up @@ -542,6 +542,13 @@
"edge_api.identities.openapi",
"environments.identities.traits.openapi",
],
"PREPROCESSING_HOOKS": [
"api.openapi.preprocessing_filter_spec",
],
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Default preprocessing hook dropped when overriding PREPROCESSING_HOOKS

Medium Severity

The POSTPROCESSING_HOOKS correctly re-includes the drf-spectacular default drf_spectacular.hooks.postprocess_schema_enums, but PREPROCESSING_HOOKS does not re-include the default drf_spectacular.hooks.preprocess_exclude_path_format. That default hook filters out format-suffixed path duplicates (e.g. .json/.api variants). Overriding the setting with only the new preprocessing_filter_spec hook silently drops it, potentially causing duplicate or format-suffixed paths to appear in the generated schema.

Fix in Cursor Fix in Web

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The default value for PREPROCESSING_HOOKS in drf-spectacular is [], preprocess_exclude_path_format is only listed as an example in the docs, not included by default. So we're not dropping any existing behaviour here.

"POSTPROCESSING_HOOKS": [
"drf_spectacular.hooks.postprocess_schema_enums",
"api.openapi.postprocessing_assign_tags",
],
"ENUM_NAME_OVERRIDES": {
# Overrides to use specific schema names for fields named "type".
# If this is not set, drf-spectacular will generate schema names like "Type975Enum".
Expand Down
141 changes: 140 additions & 1 deletion api/tests/unit/api/test_unit_openapi.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,16 @@
from typing import Any

import pytest
from drf_spectacular.generators import SchemaGenerator
from drf_spectacular.openapi import AutoSchema
from typing_extensions import TypedDict

from api.openapi import TypedDictSchemaExtension
from api.openapi import (
TAGS,
TypedDictSchemaExtension,
postprocessing_assign_tags,
preprocessing_filter_spec,
)


def test_typeddict_schema_extension__renders_expected() -> None:
Expand Down Expand Up @@ -86,6 +94,137 @@ class ResponseModel(TypedDict):
}


@pytest.mark.parametrize(
"path, expected_tag",
[
("/api/v1/organisations/", "Organisations"),
("/api/v1/organisations/{id}/groups/", "Organisations"),
("/api/v1/projects/{id}/", "Projects"),
("/api/v1/environments/{api_key}/", "Environments"),
("/api/v1/projects/{id}/features/", "Features"),
("/api/v1/flags/{feature_id}/multivariate-options/", "Features"),
("/api/v1/environments/{api_key}/featurestates/{id}/", "Feature states"),
("/api/v1/environment-feature-versions/{id}/", "Feature states"),
("/api/v1/environments/{api_key}/identities/{id}/", "Identities"),
("/api/v1/environments/{api_key}/edge-identities/{id}/", "Identities"),
("/api/v1/traits/", "Identities"),
("/api/v1/segments/{id}/", "Segments"),
("/api/v1/environments/{api_key}/integrations/amplitude/{id}/", "Integrations"),
("/api/v1/projects/{id}/integrations/datadog/{id}/", "Integrations"),
("/api/v1/organisations/{id}/integrations/github/", "Integrations"),
("/api/v1/environments/{api_key}/user-permissions/{id}/", "Permissions"),
("/api/v1/projects/{id}/user-group-permissions/{id}/", "Permissions"),
("/api/v1/environments/{api_key}/webhooks/{id}/", "Webhooks"),
("/api/v1/cb-webhook/", "Webhooks"),
("/api/v1/github-webhook/", "Webhooks"),
("/api/v1/audit/", "Audit"),
("/api/v1/auth/login/", "Authentication"),
("/api/v1/users/join/{hash}/", "Authentication"),
("/api/v1/analytics/flags/", "Analytics"),
("/api/v1/metadata/fields/", "Metadata"),
("/api/v1/onboarding/request/send/", "Onboarding"),
("/api/v1/admin/dashboard/summary/", "Admin dashboard"),
],
)
def test_postprocessing_assign_tags__assigns_correct_tag(
path: str, expected_tag: str
) -> None:
# Given
result: dict[str, Any] = {
"paths": {
path: {
"get": {
"operationId": "test_op",
"tags": ["api"],
},
},
},
}

# When
postprocessing_assign_tags(result, generator=None)

# Then
assert result["paths"][path]["get"]["tags"] == [expected_tag]


def test_postprocessing_assign_tags__preserves_explicit_tags() -> None:
# Given
result: dict[str, Any] = {
"paths": {
"/api/v1/flags/": {
"get": {
"operationId": "sdk_flags",
"tags": ["sdk"],
},
},
"/api/v1/organisations/": {
"get": {
"operationId": "organisations_list",
"tags": ["mcp", "organisations"],
},
},
},
}

# When
postprocessing_assign_tags(result, generator=None)

# Then
assert result["paths"]["/api/v1/flags/"]["get"]["tags"] == ["sdk"]
assert result["paths"]["/api/v1/organisations/"]["get"]["tags"] == [
"mcp",
"organisations",
]


def test_postprocessing_assign_tags__sets_tags_list_on_result() -> None:
# Given
result: dict[str, Any] = {"paths": {}}

# When
postprocessing_assign_tags(result, generator=None)

# Then
assert result["tags"] == TAGS


def test_postprocessing_assign_tags__unmatched_path_gets_other_tag() -> None:
# Given
result: dict[str, Any] = {
"paths": {
"/api/v1/unknown-endpoint/": {
"get": {
"operationId": "unknown",
"tags": ["api"],
},
},
},
}

# When
postprocessing_assign_tags(result, generator=None)

# Then
assert result["paths"]["/api/v1/unknown-endpoint/"]["get"]["tags"] == ["Other"]


def test_preprocessing_filter_spec__removes_swagger_endpoints() -> None:
# Given
endpoints = [
("/api/v1/organisations/", "^api/v1/organisations/", "GET", None),
("/api/v1/swagger.json", "^api/v1/swagger.json", "GET", None),
("/api/v1/swagger.yaml", "^api/v1/swagger.yaml", "GET", None),
]

# When
filtered = preprocessing_filter_spec(endpoints)

# Then
assert len(filtered) == 1
assert filtered[0][0] == "/api/v1/organisations/"


def test_typeddict_schema_extension__get_name() -> None:
# Given
class MyModel(TypedDict):
Expand Down
Loading