From 9d754f72adfbde3d44ad34003cbad07fcbaedc90 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Thu, 22 Oct 2026 20:40:43 -0400 Subject: [PATCH 1/3] OpenConceptLab/ocl_online#451 | The FHIR host serves its own Swagger page of the terminology operations On the FHIR host, /swagger/ (and swagger.json, swagger.yaml) describes the FHIR terminology service alone: search and read of CodeSystem, ValueSet and ConceptMap, $lookup, $validate-code, $expand, $translate, metadata and $versions, at /fhir and the owner bases. Each operation is listed once, tagged by resource type, with its query parameters, a Parameters body for POST, and OperationOutcome errors. Writes and the slash and no-slash duplicates are left out, and Try it out calls the FHIR host itself. When FHIR has its own host, the API's Swagger leaves the FHIR paths out (they answer 404 there) and links to the FHIR page. When FHIR shares the host, the API's Swagger lists the same curated operations instead of the generated ones. The other API operations are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R8hw7w8753DDSCezrmrDA8 --- core/common/swagger.py | 29 ++++ core/fhir/swagger.py | 284 ++++++++++++++++++++++++++++++++ core/fhir/tests_swagger.py | 233 ++++++++++++++++++++++++++ core/middlewares/middlewares.py | 9 +- 4 files changed, 552 insertions(+), 3 deletions(-) create mode 100644 core/fhir/swagger.py create mode 100644 core/fhir/tests_swagger.py diff --git a/core/common/swagger.py b/core/common/swagger.py index 75e6172a3..d43579e4e 100644 --- a/core/common/swagger.py +++ b/core/common/swagger.py @@ -1,7 +1,13 @@ """Swagger helpers for OCL-specific OpenAPI generation.""" +from copy import deepcopy + +from django.conf import settings from drf_yasg.generators import EndpointEnumerator, OpenAPISchemaGenerator +from core.fhir.base import FhirViewMixin, is_fhir_path +from core.fhir.swagger import FhirSwaggerSchemaGenerator, fhir_paths, fhir_swagger_url, on_fhir_host + class OCLSwaggerEndpointEnumerator(EndpointEnumerator): """Preserve literal dollar signs when deriving OpenAPI paths from Django URL regexes.""" @@ -18,3 +24,26 @@ class OCLSwaggerSchemaGenerator(OpenAPISchemaGenerator): """Use the OCL endpoint enumerator so Swagger paths match the published URLs.""" endpoint_enumerator_class = OCLSwaggerEndpointEnumerator + + def get_endpoints(self, request): + """Replace every FHIR view entry, including malformed paths from no-slash includes.""" + return {path: endpoint for path, endpoint in super().get_endpoints(request).items() + if not is_fhir_path(path) and not issubclass(endpoint[0], FhirViewMixin)} + + def get_paths(self, endpoints, components, request, public): + """Keep the other API operations intact, adding FHIR only when both share a host.""" + paths, prefix = super().get_paths(endpoints, components, request, public) + if not settings.FHIR_SUBDOMAIN: + paths.update(fhir_paths(components)) + return paths, prefix + + def get_schema(self, request=None, public=False): + """Select the host's schema and link the API documentation to the separate FHIR service.""" + if on_fhir_host(request): + return FhirSwaggerSchemaGenerator(self.info).get_schema(request, public) + schema = super().get_schema(request, public) + if settings.FHIR_SUBDOMAIN: + schema.info = deepcopy(schema.info) + link = fhir_swagger_url(request, self.url or settings.API_BASE_URL) + schema.info.description = schema.info.get('description', '') + f'\n\n[FHIR terminology Swagger]({link}).' + return schema diff --git a/core/fhir/swagger.py b/core/fhir/swagger.py new file mode 100644 index 000000000..9eb17622b --- /dev/null +++ b/core/fhir/swagger.py @@ -0,0 +1,284 @@ +"""Curated Swagger 2.0 descriptions of the FHIR terminology reads and operations.""" +from copy import deepcopy +from urllib.parse import urlsplit, urlunsplit + +from django.conf import settings +from drf_yasg import openapi +from drf_yasg.generators import OpenAPISchemaGenerator + +from core.fhir.base import FHIR_JSON, FHIR_XML +from core.fhir.ops.expand import MAX_COUNT as EXPANSION_MAX_COUNT +from core.fhir.search import MAX_COUNT, PARAMETERS, STRING_FIELDS, STRING_LOOKUPS + +BASES = ('/fhir', '/orgs/{org}', '/users/{user}') +TAGS = ('CodeSystem', 'ValueSet', 'ConceptMap', 'Capabilities') +VERSION_SELECTION = ( + 'The default is the latest released version visible at this base; HEAD is used only when explicitly named. ' + 'An instance id at /fhir names a released version; at an owner base it is the repository mnemonic.') +SEARCH_DESCRIPTIONS = { + '_id': 'Resource id: an opaque release id at /fhir, or the repository mnemonic at an owner base.', + '_lastUpdated': 'FHIR date or dateTime, optionally prefixed by eq, ne, gt, lt, ge, le, sa or eb.', + 'url': 'Canonical URL of the resource.', + 'version': 'Business version token. Searches include released versions only, never HEAD.', + 'status': 'Publication status token: active or retired.', + 'identifier': 'Identifier token: value, system|value, |value or system|.', + 'system': 'Canonical URL of the code system (an alias of url).', + 'code': 'A code defined by the code system; code or system|code.', + 'language': 'Language token for the code system (urn:ietf:bcp:47).', + 'reference': 'Canonical code system URL referenced by the value set\'s stored default expansion.', + 'source-system': 'Canonical URL of a source system in a ConceptMap group.', + 'source-uri': 'Alias of source-system.', + 'target-system': 'Canonical URL of a target system in a ConceptMap group.', + 'target-uri': 'Alias of target-system.', +} + + +def on_fhir_host(request): + """Use the same host boundary for the schema and the FHIR middleware.""" + return bool(request and settings.FHIR_SUBDOMAIN and + request.get_host().startswith(settings.FHIR_SUBDOMAIN + '.')) + + +def fhir_swagger_url(request, api_url): + """The FHIR sibling of the API host, keeping the request's scheme and port.""" + url = urlsplit(request.build_absolute_uri('/') if request else api_url) + host = settings.FHIR_SUBDOMAIN + '.' + (url.netloc.partition('.')[2] or url.netloc) + return urlunsplit((url.scheme, host, '/swagger/', '', '')) + + +def query(name, description, type_=openapi.TYPE_STRING, **kwargs): + """A typed query parameter; complex FHIR values are described in the POST body instead.""" + return openapi.Parameter(name, openapi.IN_QUERY, type=type_, description=description, **kwargs) + + +FORMAT = query('_format', 'Response format; overrides Accept.', enum=['json', 'xml']) +CODE = query('code', 'Code to look up, validate or translate; alternatively supply coding.') +CODING = query('coding', 'A system|code token, instead of separate code and system parameters.') +SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from.') +VERSION = query('version', 'Business version of the code system; defaults to the latest released version.') +URL = query('url', 'Canonical URL or OCL path of the target resource; may include |version.') +VALUE_SET_VERSION = query('valueSetVersion', 'Business version of the value set.') +LANGUAGE = query('displayLanguage', 'Preferred language for the returned display, when a name in it exists.') +DISPLAY = query('display', 'Display to validate against the concept\'s names, ignoring case.') +OPERATION_PARAMETERS = { + ('CodeSystem', '$lookup'): [SYSTEM, VERSION, CODE, CODING, LANGUAGE], + ('CodeSystem', '$validate-code'): [URL, VERSION, CODE, CODING, DISPLAY, LANGUAGE], + ('ValueSet', '$validate-code'): [ + URL, VALUE_SET_VERSION, SYSTEM, query('systemVersion', 'Business version of the input code system.'), + CODE, CODING, DISPLAY, LANGUAGE], + ('ValueSet', '$expand'): [ + URL, VALUE_SET_VERSION, + query('filter', 'Each word must prefix a word in the code or one of its names, ignoring case.'), LANGUAGE, + query('includeDesignations', 'Include every name as a designation.', openapi.TYPE_BOOLEAN), + query('activeOnly', 'Exclude retired concepts.', openapi.TYPE_BOOLEAN), + query('offset', 'Number of matching members to skip.', openapi.TYPE_INTEGER, minimum=0, default=0), + query('count', 'Page size, capped at 1,000; zero returns only the total.', openapi.TYPE_INTEGER, + minimum=0, maximum=EXPANSION_MAX_COUNT, default=EXPANSION_MAX_COUNT), + query('includeDefinition', 'Only false is supported; definitions are not included.', openapi.TYPE_BOOLEAN, + enum=[False]), + query('excludePostCoordinated', 'Only false is supported; stored expansions do not record this distinction.', + openapi.TYPE_BOOLEAN, enum=[False]), + query('excludeNested', 'Either boolean is accepted; stored expansions are flat.', openapi.TYPE_BOOLEAN), + query('excludeNotForUI', 'Either boolean is accepted; stored expansions have no abstract codes.', + openapi.TYPE_BOOLEAN)], + ('ConceptMap', '$translate'): [ + URL, query('conceptMapVersion', 'Business version of the map; without url, selects the source\'s map version.'), + CODE, SYSTEM, + query('version', 'Input code system version; omitted means no mapping-side version filter. ' + 'Without url in forward translation, also selects the map version unless conceptMapVersion ' + 'is given.'), CODING, + query('source', 'Input value set scope for forward translation; ignored when reverse=true.'), + query('target', 'Input value set scope for reverse translation; ignored in forward translation.'), + query('targetsystem', 'Restrict matches to this code system; in reverse, the system of the codes sought.'), + query('reverse', 'Translate from the target side of the mappings.', openapi.TYPE_BOOLEAN, default=False)], +} +OPERATION_DESCRIPTIONS = { + ('CodeSystem', '$lookup'): ( + 'Look up a code', 'Returns the code\'s display, names and inactive flag, including retired concepts. ' + 'A code and, at type level, its system are required (or supply coding).', (400, 404, 422)), + ('CodeSystem', '$validate-code'): ( + 'Validate a code in a code system', + 'Checks the code and optional display in url (or coding.system). A code and a target system are required. ' + 'An unknown code or system returns Parameters with result=false; retired codes are valid with a warning.', + (400,)), + ('ValueSet', '$validate-code'): ( + 'Validate a code in a value set', + 'Checks membership in the stored default expansion and an optional display. A code and its system are ' + 'required (or supply coding). Without url at type level, checks the code system alone. ' + 'A failed validation returns Parameters with result=false.', (400, 422)), + ('ValueSet', '$expand'): ( + 'Read a stored value set expansion', + 'Returns a page of the stored default expansion; never computes or saves an expansion. ' + 'At type level url is required. count is capped at 1,000. Missing, unfinished or incomplete stored ' + 'expansions return 422. Inline value sets, contexts, dates, selected designations and code system version ' + 'overrides are not supported.', (400, 422)), + ('ConceptMap', '$translate'): ( + 'Translate a code', + 'Returns live mappings from the named ConceptMap. Without url, uses the input system\'s own mappings ' + 'in forward translation, or targetsystem\'s mappings in reverse; without targetsystem, reverse has no match. ' + 'A code and system are required (or supply coding). Supports reverse translation and a target system filter. ' + 'No match returns Parameters with result=false. A POSTed codeableConcept uses its first coding.', + (400, 404, 422)), +} + + +def resource_schema(*resource_types): + """Identify the FHIR response type without duplicating the complete FHIR resource definitions.""" + return openapi.Schema(type=openapi.TYPE_OBJECT, required=['resourceType'], properties={ + 'resourceType': openapi.Schema(type=openapi.TYPE_STRING, enum=list(resource_types)), + }) + + +def parameters_schema(components): + """One shared Parameters body, including the structured alternatives to query-string coding.""" + definitions = components.with_scope(openapi.SCHEMA_DEFINITIONS) + if not definitions.has('FhirParameters'): + schema = resource_schema('Parameters') + schema.properties['parameter'] = openapi.Schema(type=openapi.TYPE_ARRAY, items=openapi.Schema( + type=openapi.TYPE_OBJECT, required=['name'], properties={ + 'name': openapi.Schema(type=openapi.TYPE_STRING), + 'valueString': openapi.Schema(type=openapi.TYPE_STRING), + 'valueCode': openapi.Schema(type=openapi.TYPE_STRING), + 'valueUri': openapi.Schema(type=openapi.TYPE_STRING), + 'valueBoolean': openapi.Schema(type=openapi.TYPE_BOOLEAN), + 'valueInteger': openapi.Schema(type=openapi.TYPE_INTEGER), + 'valueCoding': openapi.Schema(type=openapi.TYPE_OBJECT, properties={ + key: openapi.Schema(type=openapi.TYPE_STRING) for key in ('system', 'code', 'version', 'display')}), + 'valueCodeableConcept': openapi.Schema(type=openapi.TYPE_OBJECT, properties={ + 'coding': openapi.Schema(type=openapi.TYPE_ARRAY, items=openapi.Schema(type=openapi.TYPE_OBJECT))}), + })) + definitions.set('FhirParameters', schema) + return openapi.SchemaRef(definitions, 'FhirParameters') + + +def operation(summary, description, parameters, response_type, errors=(400, 422)): + """A read with FHIR response media types and OperationOutcome errors; a token shows private content.""" + responses = {'200': openapi.Response(response_type, schema=resource_schema(response_type))} + errors_by_status = { + 400: 'Invalid or unsupported input.', 404: 'Resource or code not found or not visible at this base.', + 422: 'The request cannot be processed or exceeds the operation\'s processing bound.', + } + responses.update({str(code): openapi.Response(errors_by_status[code], schema=resource_schema('OperationOutcome')) + for code in errors}) + return openapi.Operation( + operation_id='', summary=summary, description=description, parameters=[*parameters, FORMAT], + responses=openapi.Responses(responses), produces=[FHIR_JSON, FHIR_XML]) + + +def search_operation(resource_type): + """Take the supported names and modifiers from the search parser itself.""" + parameters = [query(name, SEARCH_DESCRIPTIONS[name]) for name in PARAMETERS[resource_type]] + modifiers = {'': 'Starts with, ignoring case and accents.', 'exact': 'Exact, including case and accents.', + 'contains': 'Contains, ignoring case and accents.'} + parameters.extend(query(f'{name}:{modifier}' if modifier else name, f'{name}: {modifiers[modifier]}') + for name in STRING_FIELDS for modifier in STRING_LOOKUPS) + parameters.extend([ + query('_count', 'Page size, capped at 100; zero returns only the total.', openapi.TYPE_INTEGER, + minimum=0, maximum=MAX_COUNT, default=MAX_COUNT), + query('page', 'Page number, starting at 1.', openapi.TYPE_INTEGER, minimum=1, default=1), + ]) + return operation(f'Search {resource_type} resources', + 'Returns a searchset Bundle of visible released versions, never HEAD. Entries are summaries ' + 'tagged SUBSETTED. Commas separate alternatives; repeated parameters must all match. ' + 'Unsupported parameters or modifiers return 400.', parameters, 'Bundle') + + +def resource_operations(resource_type, base): + """The type's search, read and supported terminology operations, once per canonical suffix.""" + read_parameters = [] if base == '/fhir' else [query('version', 'Released business version; never HEAD.')] + read_description = ( + 'The opaque id selects one released version.' if base == '/fhir' else + 'The id is the repository mnemonic. Returns the latest visible release unless version is named.') + paths = { + f'/{resource_type}': {'get': search_operation(resource_type)}, + f'/{resource_type}/{{id}}': {'get': operation( + f'Read a {resource_type}', read_description + ' HEAD is never returned. A cut read is tagged SUBSETTED.', + read_parameters, resource_type, (400, 404, 422))}, + } + for (resource, name), (summary, description, errors) in OPERATION_DESCRIPTIONS.items(): + if resource != resource_type: + continue + for instance in (False, True) if resource != 'ConceptMap' else (False,): + suffix = f'/{resource}/' + ('{id}/' if instance else '') + name + operation_errors = tuple(sorted(set(errors) | ({404} if instance else set()))) + documented = operation(summary, description + ' ' + VERSION_SELECTION, + OPERATION_PARAMETERS[(resource, name)], + 'ValueSet' if name == '$expand' else 'Parameters', operation_errors) + paths[suffix] = {'get': documented, 'post': deepcopy(documented)} + return paths + + +def capability_operations(): + """The metadata variants share a route; $versions only supports GET.""" + metadata = operation('Read service capabilities', + 'Returns CapabilityStatement, or TerminologyCapabilities for mode=terminology.', + [query('mode', 'Use terminology to list the code systems and versions visible at this base.', + enum=['terminology'])], 'CapabilityStatement') + metadata.responses['200'] = openapi.Response( + 'CapabilityStatement; TerminologyCapabilities when mode=terminology.', + schema=resource_schema('CapabilityStatement', 'TerminologyCapabilities')) + return { + '/metadata': {'get': metadata}, + '/$versions': {'get': operation('List supported FHIR versions', + 'Returns the supported FHIR versions and the default (4.0 for R4).', + [], 'Parameters', (400,))}, + } + + +def path_parameters(path): + """Document the FHIR spelling of the route variables, independent of Django's lookup kwarg names.""" + descriptions = { + 'org': 'Organization identifier.', 'user': 'User identifier.', + 'id': 'Opaque release id at /fhir; repository mnemonic at an owner base.', + } + return [openapi.Parameter(name, openapi.IN_PATH, type=openapi.TYPE_STRING, required=True, description=description) + for name, description in descriptions.items() if '{' + name + '}' in path] + + +def fhir_paths(components): + """One source of paths for the dedicated FHIR schema and the combined API schema.""" + paths = {} + body = parameters_schema(components) + for tag in TAGS: + for base in BASES: + templates = capability_operations() if tag == 'Capabilities' else resource_operations(tag, base) + for suffix, methods in templates.items(): + path = base + suffix + for method, documented in methods.items(): + documented.operation_id = 'fhir_' + '_'.join( + path.strip('/').replace('{', '').replace('}', '').replace('$', '').replace('-', '_').split('/') + ) + '_' + method + documented.tags = [tag] + documented.parameters = deepcopy(documented.parameters) + path_parameters(path) + if method == 'post': + documented.consumes = [FHIR_JSON] + documented.parameters.append(openapi.Parameter( + 'parameters', openapi.IN_BODY, schema=body, + description='FHIR Parameters: use the query input names with one value[x] per parameter. ' + 'coding may use valueCoding; codeableConcept may use valueCodeableConcept ' + '(one coding for lookup/validation). Body values take precedence over query ' + 'values. _format stays in the query string.')) + paths[path] = openapi.PathItem(**methods) + return openapi.Paths(paths) + + +class FhirSwaggerSchemaGenerator(OpenAPISchemaGenerator): + """Generate the curated service schema without inspecting inherited resource write methods.""" + + def get_schema(self, request=None, public=False): # pylint: disable=unused-argument + components = self.reference_resolver_class(openapi.SCHEMA_DEFINITIONS, force_init=True) + paths = fhir_paths(components) + api_url = settings.API_BASE_URL.rstrip('/') + '/swagger/' + info = openapi.Info( + title='OCL FHIR terminology service (R4)', default_version='4.0.1', + description='FHIR R4 terminology in JSON and XML. Serves the SVCM transactions Query Value Set (ITI-95), ' + 'Query Code System (ITI-96), Expand Value Set (ITI-97), Lookup Code (ITI-98), ' + 'Validate Code (ITI-99), Query Concept Map (ITI-100) and Translate Code (ITI-101). ' + 'Bases: /fhir, /orgs/{org} and /users/{user}. The global base shows public resources; ' + 'owner bases show resources visible to the caller. ' + f'[OCL API Swagger]({api_url}) describes the rest of OCL.') + security_definitions = self.get_security_definitions() + return openapi.Swagger( + info=info, paths=paths, security_definitions=security_definitions, + security=self.get_security_requirements(security_definitions), + _url=request.build_absolute_uri('/') if request else self.url, _prefix='', **dict(components)) diff --git a/core/fhir/tests_swagger.py b/core/fhir/tests_swagger.py new file mode 100644 index 000000000..f06db251c --- /dev/null +++ b/core/fhir/tests_swagger.py @@ -0,0 +1,233 @@ +"""The terminology schema describes the routed reads and keeps the other API operations unchanged.""" +import re +from pathlib import Path +from tempfile import TemporaryDirectory + +from django.conf import settings +from django.http import HttpResponse +from django.test import RequestFactory, SimpleTestCase, override_settings +from django.urls import resolve +from drf_yasg import openapi +from drf_yasg.generators import OpenAPISchemaGenerator +from rest_framework.request import Request +from rest_framework.test import APIClient, APITestCase +from whitenoise.middleware import WhiteNoiseMiddleware + +from core.common.swagger import OCLSwaggerEndpointEnumerator, OCLSwaggerSchemaGenerator +from core.fhir.base import FhirViewMixin, is_fhir_path +from core.fhir.search import PARAMETERS, STRING_FIELDS, STRING_LOOKUPS +from core.fhir.swagger import FhirSwaggerSchemaGenerator +from core.middlewares.middlewares import FhirMiddleware +from core.urls import api_info + +BASES = ('/fhir', '/orgs/{org}', '/users/{user}') +OPERATIONS = { + 'CodeSystem': ('/$lookup', '/{id}/$lookup', '/$validate-code', '/{id}/$validate-code'), + 'ValueSet': ('/$validate-code', '/{id}/$validate-code', '/$expand', '/{id}/$expand'), + 'ConceptMap': ('/$translate',), +} +MIDDLEWARE = ['core.middlewares.middlewares.FhirMiddleware'] +SERVICE_MIDDLEWARE = tuple(settings.MIDDLEWARE) + + +def expected_operations(): + """An independent pin of the documented methods and paths at all three bases.""" + expected = {} + for base in BASES: + for resource, operations in OPERATIONS.items(): + for suffix in ('', '/{id}'): + expected[('get', f'{base}/{resource}{suffix}')] = resource + for suffix in operations: + for method in ('get', 'post'): + expected[(method, f'{base}/{resource}{suffix}')] = resource + for suffix in ('/metadata', '/$versions'): + expected[('get', base + suffix)] = 'Capabilities' + return expected + + +def operations_in(schema): + """Index actual schema operations without counting path-level parameters.""" + return {(method, path): operation for path, item in schema['paths'].items() + for method, operation in item.items() if method in openapi.PathItem.OPERATION_NAMES} + + +class FhirSwaggerTest(SimpleTestCase): + """Schema generation is independent of resource data and serializer introspection.""" + + def setUp(self): + self.schema = FhirSwaggerSchemaGenerator(api_info).get_schema(public=True) + + def test_exact_operations_and_routed_methods(self): + actual, expected = operations_in(self.schema), expected_operations() + self.assertEqual(set(actual), set(expected)) + self.assertEqual(len(actual), 78) + identifiers = [] + for (method, path), operation in actual.items(): + self.assertFalse(path.endswith('/'), path) + self.assertEqual(operation['tags'], [expected[(method, path)]]) + self.assertTrue(operation['summary']) + self.assertNotIn('\n', operation['summary']) + self.assertTrue(operation['description']) + self.assertNotIn('security', operation) # the schema's: a token shows private content at owner bases + identifiers.append(operation['operationId']) + concrete = path.replace('{org}', 'Acme').replace('{user}', 'jo').replace('{id}', '123') + view_class = resolve(concrete).func.view_class + self.assertTrue(issubclass(view_class, FhirViewMixin), path) + self.assertIn(method, view_class.http_method_names, path) + self.assertTrue(callable(getattr(view_class, method, None)), path) + parameters = operation['parameters'] + self.assertEqual({p['name'] for p in parameters if p['in'] == 'path'}, set(re.findall(r'{(\w+)}', path))) + self.assertEqual(len({(p['in'], p['name']) for p in parameters}), len(parameters)) + self.assertIn('_format', [p['name'] for p in parameters]) + self.assertEqual(len(identifiers), len(set(identifiers))) + + def test_search_parameters_match_the_search_parser(self): + for base in BASES: + for resource in OPERATIONS: + parameters = self.schema['paths'][f'{base}/{resource}']['get']['parameters'] + query = {p['name']: p for p in parameters if p['in'] == 'query'} + names = {f'{name}:{modifier}' if modifier else name + for name in STRING_FIELDS for modifier in STRING_LOOKUPS} + self.assertEqual(set(query), set(PARAMETERS[resource]) | names | {'_count', 'page', '_format'}) + self.assertEqual(query['_count']['type'], 'integer') + self.assertEqual(query['_count']['maximum'], 100) + self.assertEqual(query['_format']['enum'], ['json', 'xml']) + + def test_operation_parameters_and_bodies(self): + expected = { + 'CodeSystem/$lookup': {'system', 'version', 'code', 'coding', 'displayLanguage'}, + 'CodeSystem/$validate-code': {'url', 'version', 'code', 'coding', 'display', 'displayLanguage'}, + 'ValueSet/$validate-code': { + 'url', 'valueSetVersion', 'system', 'systemVersion', 'code', 'coding', 'display', 'displayLanguage'}, + 'ValueSet/$expand': { + 'url', 'valueSetVersion', 'filter', 'displayLanguage', 'includeDesignations', 'activeOnly', + 'offset', 'count', 'includeDefinition', 'excludePostCoordinated', 'excludeNested', 'excludeNotForUI'}, + 'ConceptMap/$translate': { + 'url', 'conceptMapVersion', 'code', 'system', 'version', 'coding', 'source', 'target', + 'targetsystem', 'reverse'}, + } + for path, names in expected.items(): + for base in BASES: + for method in ('get', 'post'): + operation = self.schema['paths'][f'{base}/{path}'][method] + parameters = operation['parameters'] + query = {p['name']: p for p in parameters if p['in'] == 'query'} + self.assertEqual(set(query), names | {'_format'}) + self.assertTrue(all('type' in parameter for parameter in query.values())) + bodies = [p for p in parameters if p['in'] == 'body'] + self.assertEqual(len(bodies), int(method == 'post')) + if bodies: + self.assertEqual(operation['consumes'], ['application/fhir+json']) + self.assertEqual(bodies[0]['schema']['$ref'], '#/definitions/FhirParameters') + expand = self.schema['paths']['/fhir/ValueSet/$expand']['get']['parameters'] + count = next(p for p in expand if p['name'] == 'count') + self.assertEqual((count['type'], count['maximum']), ('integer', 1000)) + + def test_tags_order_responses_and_security_definitions(self): + for tag in (*OPERATIONS, 'Capabilities'): + paths = [path for (_, path), operation in operations_in(self.schema).items() if operation['tags'] == [tag]] + order = [next(index for index, base in enumerate(BASES) if path.startswith(base + '/')) for path in paths] + self.assertEqual(order, sorted(order)) + self.assertEqual(set(self.schema['securityDefinitions']), {'Basic', 'Token'}) + self.assertEqual({name for requirement in self.schema['security'] for name in requirement}, {'Basic', 'Token'}) + for (_, path), operation in operations_in(self.schema).items(): + success = operation['responses']['200'] + self.assertTrue(success['schema']) + for code, response in operation['responses'].items(): + if code != '200': + self.assertEqual(response['schema']['properties']['resourceType']['enum'], ['OperationOutcome']) + if path.endswith('/metadata'): + self.assertIn('TerminologyCapabilities', success['description']) + + +@override_settings(FHIR_SUBDOMAIN='fhir', MIDDLEWARE=MIDDLEWARE, ALLOWED_HOSTS=['*']) +class FhirSwaggerHostTest(SimpleTestCase): + """The FHIR host serves its own documentation without opening other API paths.""" + + def test_schema_urls_use_the_request_host_and_forwarded_scheme(self): + for path in ('/swagger/?format=openapi', '/swagger.json', '/swagger.yaml'): + response = APIClient().get(path, HTTP_HOST='fhir.example.org:8443', HTTP_X_FORWARDED_PROTO='https') + self.assertEqual(response.status_code, 200, path) + self.assertEqual(set(operations_in(response.data)), set(expected_operations())) + self.assertEqual(response.data['host'], 'fhir.example.org:8443') + self.assertEqual(response.data['schemes'], ['https']) + self.assertEqual(response.data['basePath'], '/') + self.assertIn('FHIR terminology', response.data['info']['title']) + + def test_only_the_documentation_paths_open_on_the_fhir_host(self): + response = APIClient().get('/swagger/', HTTP_HOST='fhir.example.org', HTTP_ACCEPT='text/html') + self.assertEqual(response.status_code, 200) + middleware = FhirMiddleware(lambda request: HttpResponse('allowed')) + for path in ('/swagger/', '/swagger.json', '/swagger.yaml', '/version/'): + request = RequestFactory().get(path, HTTP_HOST='fhir.example.org') + self.assertEqual(middleware(request).content, b'allowed', path) + for path in ('/orgs/', '/redoc/', '/swagger', '/swagger/other', '/swagger.json/extra', '/swagger.yaml/'): + response = APIClient().get(path, HTTP_HOST='fhir.example.org') + self.assertEqual(response.status_code, 404, path) + self.assertEqual(response.json()['resourceType'], 'OperationOutcome', path) + + def test_the_ui_files_are_served_ahead_of_the_host_check(self): + """Static files are served before FhirMiddleware, so the Swagger UI's files load on the FHIR host.""" + static = 'whitenoise.middleware.WhiteNoiseMiddleware' + if static in SERVICE_MIDDLEWARE: + self.assertLess(SERVICE_MIDDLEWARE.index(static), + SERVICE_MIDDLEWARE.index('core.middlewares.middlewares.FhirMiddleware')) + with TemporaryDirectory() as directory: + target = Path(directory) / 'drf-yasg' / 'style.css' + target.parent.mkdir(parents=True) + target.write_text('body {}') + with self.settings(STATIC_ROOT=directory, STATIC_URL='/static/', WHITENOISE_AUTOREFRESH=True, + WHITENOISE_USE_FINDERS=False): + middleware = WhiteNoiseMiddleware(FhirMiddleware( + lambda request: HttpResponse('not static', status=404))) + response = middleware(RequestFactory().get('/static/drf-yasg/style.css', HTTP_HOST='fhir.example.org')) + self.assertEqual(response.status_code, 200) + response.close() + + +class OriginalSchemaGenerator(OpenAPISchemaGenerator): + """The API generator before FHIR curation, for a comparison of non-FHIR operations.""" + endpoint_enumerator_class = OCLSwaggerEndpointEnumerator + + +@override_settings(MIDDLEWARE=MIDDLEWARE, ALLOWED_HOSTS=['*']) +class ApiSwaggerTest(APITestCase): + """Compare the complete API schema using real resolver endpoints and the original inspector.""" + + def schema(self): + response = self.client.get('/swagger/?format=openapi', HTTP_HOST='api.example.org', + HTTP_X_FORWARDED_PROTO='https') + self.assertEqual(response.status_code, 200) + return response.data + + @override_settings(FHIR_SUBDOMAIN='fhir') + def test_separate_host_omits_fhir_and_links_to_its_swagger(self): + schema = self.schema() + self.assertFalse(any(is_fhir_path(path) for path in schema['paths'])) + self.assertFalse(any('CodeSystem' in path or 'ValueSet' in path or 'ConceptMap' in path + for path in schema['paths'])) + self.assertIn('https://fhir.example.org/swagger/', schema['info']['description']) + self.assertNotIn('fhir.example.org', api_info.description) + + @override_settings(FHIR_SUBDOMAIN=None) + def test_same_host_includes_only_curated_fhir(self): + schema = self.schema() + fhir = {key: operation for key, operation in operations_in(schema).items() if is_fhir_path(key[1])} + curated = operations_in(FhirSwaggerSchemaGenerator(api_info).get_schema(public=True)) + self.assertEqual(fhir, curated) + self.assertFalse(any('CodeSystem$' in path or 'ValueSet$' in path or 'ConceptMap$' in path + for path in schema['paths'])) + + def test_non_fhir_paths_and_operations_are_unchanged(self): + request = Request(RequestFactory().get('/swagger/', HTTP_HOST='api.example.org')) + original = OriginalSchemaGenerator(api_info) + endpoints = {path: endpoint for path, endpoint in original.get_endpoints(request).items() + if not issubclass(endpoint[0], FhirViewMixin) and not is_fhir_path(path)} + original_schema = original.get_schema(request, public=True) + expected = {path: item for path, item in original_schema.paths.items() if path in endpoints} + self.assertTrue(expected) + for subdomain in ('fhir', None): + with self.settings(FHIR_SUBDOMAIN=subdomain): + schema = OCLSwaggerSchemaGenerator(api_info).get_schema(request, public=True) + actual = {path: item for path, item in schema.paths.items() if not is_fhir_path(path)} + self.assertEqual(actual, dict(expected)) diff --git a/core/middlewares/middlewares.py b/core/middlewares/middlewares.py index 15eb94f7b..635afae1c 100644 --- a/core/middlewares/middlewares.py +++ b/core/middlewares/middlewares.py @@ -19,6 +19,7 @@ from core.common.utils import set_current_user, set_request_url from core.fhir.base import FHIR_JSON, is_fhir_path, is_outcome_response, outcome_response, response_json from core.fhir.outcomes import issue, operation_outcome, outcome_for_status +from core.fhir.swagger import on_fhir_host from core.services.analytics_event_emitter import AnalyticsEventEmitter from core.services.auth.core import AuthService @@ -298,14 +299,16 @@ class FhirMiddleware(BaseMiddleware): Keeps FHIR on the FHIR host when FHIR_SUBDOMAIN is set: FHIR resource paths 404 elsewhere, and other paths 404 on the FHIR host, each with an OperationOutcome. """ + documentation_paths = ('/swagger/', '/swagger.json', '/swagger.yaml') # the FHIR host's own Swagger def __call__(self, request): fhir_path = is_fhir_path(request.path) if settings.FHIR_SUBDOMAIN: - on_fhir_host = request.get_host().startswith(settings.FHIR_SUBDOMAIN + '.') - if on_fhir_host and not fhir_path and request.path.strip('/').split('/')[0] != 'version': + fhir_host = on_fhir_host(request) + if fhir_host and not fhir_path and request.path.strip('/').split('/')[0] != 'version' and \ + request.path not in self.documentation_paths: return self.not_found(request, 'This path is not part of the FHIR API.') - if not on_fhir_host and fhir_path: + if not fhir_host and fhir_path: return self.not_found(request, 'The FHIR API is served on its own host.') return self.get_response(request) From e06ba677e6160a708347232db57316c6b0869a3c Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Thu, 22 Oct 2026 21:15:01 -0400 Subject: [PATCH 2/3] OpenConceptLab/ocl_online#451 | FHIR Swagger: say which inputs a coding replaces When a coding or codeableConcept is given, the operations don't read the separate inputs it stands for: code and system for $lookup (version still selects the code system version), code and display for CodeSystem $validate-code (url and version still name the code system), code, system, systemVersion and display for ValueSet $validate-code, and code, system and version for $translate. Each operation's coding parameter now names them, and each of those inputs says it isn't read alongside a coding. A GET coding token carries no version or display; a POSTed valueCoding does. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R8hw7w8753DDSCezrmrDA8 --- core/fhir/swagger.py | 33 ++++++++++++++++++++++++--------- core/fhir/tests_swagger.py | 12 ++++++++++++ 2 files changed, 36 insertions(+), 9 deletions(-) diff --git a/core/fhir/swagger.py b/core/fhir/swagger.py index 9eb17622b..be1c16777 100644 --- a/core/fhir/swagger.py +++ b/core/fhir/swagger.py @@ -51,21 +51,35 @@ def query(name, description, type_=openapi.TYPE_STRING, **kwargs): return openapi.Parameter(name, openapi.IN_QUERY, type=type_, description=description, **kwargs) +NOT_WITH_CODING = ' Not read when coding or codeableConcept is given.' FORMAT = query('_format', 'Response format; overrides Accept.', enum=['json', 'xml']) -CODE = query('code', 'Code to look up, validate or translate; alternatively supply coding.') -CODING = query('coding', 'A system|code token, instead of separate code and system parameters.') -SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from.') +CODE = query('code', 'Code to look up, validate or translate; alternatively supply coding.' + NOT_WITH_CODING) +SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from.' + NOT_WITH_CODING) VERSION = query('version', 'Business version of the code system; defaults to the latest released version.') URL = query('url', 'Canonical URL or OCL path of the target resource; may include |version.') VALUE_SET_VERSION = query('valueSetVersion', 'Business version of the value set.') LANGUAGE = query('displayLanguage', 'Preferred language for the returned display, when a name in it exists.') -DISPLAY = query('display', 'Display to validate against the concept\'s names, ignoring case.') +DISPLAY = query('display', 'Display to validate against the concept\'s names, ignoring case.' + NOT_WITH_CODING) + + +def coding(replaces, applies=''): + """The coding input of one operation: a system|code token here, a full Coding in a POSTed body.""" + return query('coding', f'A system|code token. It replaces {replaces}.{applies} A GET token carries no version ' + 'or display: send those inside a POSTed valueCoding.') + + OPERATION_PARAMETERS = { - ('CodeSystem', '$lookup'): [SYSTEM, VERSION, CODE, CODING, LANGUAGE], - ('CodeSystem', '$validate-code'): [URL, VERSION, CODE, CODING, DISPLAY, LANGUAGE], + ('CodeSystem', '$lookup'): [ + SYSTEM, VERSION, CODE, coding('code and system', ' version still selects the code system version.'), + LANGUAGE], + ('CodeSystem', '$validate-code'): [ + URL, VERSION, CODE, + coding('code and display', ' url and version still name the code system validated against.'), + DISPLAY, LANGUAGE], ('ValueSet', '$validate-code'): [ - URL, VALUE_SET_VERSION, SYSTEM, query('systemVersion', 'Business version of the input code system.'), - CODE, CODING, DISPLAY, LANGUAGE], + URL, VALUE_SET_VERSION, SYSTEM, + query('systemVersion', 'Business version of the input code system.' + NOT_WITH_CODING), + CODE, coding('code, system, systemVersion and display'), DISPLAY, LANGUAGE], ('ValueSet', '$expand'): [ URL, VALUE_SET_VERSION, query('filter', 'Each word must prefix a word in the code or one of its names, ignoring case.'), LANGUAGE, @@ -86,7 +100,8 @@ def query(name, description, type_=openapi.TYPE_STRING, **kwargs): CODE, SYSTEM, query('version', 'Input code system version; omitted means no mapping-side version filter. ' 'Without url in forward translation, also selects the map version unless conceptMapVersion ' - 'is given.'), CODING, + 'is given.' + NOT_WITH_CODING), + coding('code, system and version'), query('source', 'Input value set scope for forward translation; ignored when reverse=true.'), query('target', 'Input value set scope for reverse translation; ignored in forward translation.'), query('targetsystem', 'Restrict matches to this code system; in reverse, the system of the codes sought.'), diff --git a/core/fhir/tests_swagger.py b/core/fhir/tests_swagger.py index f06db251c..fb59820da 100644 --- a/core/fhir/tests_swagger.py +++ b/core/fhir/tests_swagger.py @@ -119,6 +119,18 @@ def test_operation_parameters_and_bodies(self): if bodies: self.assertEqual(operation['consumes'], ['application/fhir+json']) self.assertEqual(bodies[0]['schema']['$ref'], '#/definitions/FhirParameters') + # a coding replaces these inputs (core.fhir.ops: coded_input, translate.read_request); each says so + replaced = { + 'CodeSystem/$lookup': {'code', 'system'}, + 'CodeSystem/$validate-code': {'code', 'display'}, + 'ValueSet/$validate-code': {'code', 'system', 'systemVersion', 'display'}, + 'ConceptMap/$translate': {'code', 'system', 'version'}, + } + for path, names in replaced.items(): + query = {p['name']: p for p in self.schema['paths'][f'/fhir/{path}']['get']['parameters']} + for name in names: + self.assertIn(name, query['coding']['description'], (path, name)) + self.assertIn('Not read when coding', query[name]['description'], (path, name)) expand = self.schema['paths']['/fhir/ValueSet/$expand']['get']['parameters'] count = next(p for p in expand if p['name'] == 'count') self.assertEqual((count['type'], count['maximum']), ('integer', 1000)) From 6bf3ea6cba639327ab66d18bc4abb28d1521aee7 Mon Sep 17 00:00:00 2001 From: Jonathan Payne Date: Thu, 22 Oct 2026 21:21:33 -0400 Subject: [PATCH 3/3] OpenConceptLab/ocl_online#451 | FHIR Swagger: $lookup's system still names an instance alongside a coding At the type level, $lookup takes the code system from a coding, so a separate system has no effect; at an instance, a separate system must still name that code system, as without a coding. Its description now says so instead of saying it isn't read. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01R8hw7w8753DDSCezrmrDA8 --- core/fhir/swagger.py | 7 ++++++- core/fhir/tests_swagger.py | 26 ++++++++++++++++---------- 2 files changed, 22 insertions(+), 11 deletions(-) diff --git a/core/fhir/swagger.py b/core/fhir/swagger.py index be1c16777..282601a65 100644 --- a/core/fhir/swagger.py +++ b/core/fhir/swagger.py @@ -56,6 +56,9 @@ def query(name, description, type_=openapi.TYPE_STRING, **kwargs): CODE = query('code', 'Code to look up, validate or translate; alternatively supply coding.' + NOT_WITH_CODING) SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from.' + NOT_WITH_CODING) VERSION = query('version', 'Business version of the code system; defaults to the latest released version.') +LOOKUP_SYSTEM = query('system', 'Canonical URL or OCL path of the code system the code is from. At an instance it ' + 'must name that code system. With coding at the type level, the coding\'s system ' + 'is used instead.') URL = query('url', 'Canonical URL or OCL path of the target resource; may include |version.') VALUE_SET_VERSION = query('valueSetVersion', 'Business version of the value set.') LANGUAGE = query('displayLanguage', 'Preferred language for the returned display, when a name in it exists.') @@ -70,7 +73,9 @@ def coding(replaces, applies=''): OPERATION_PARAMETERS = { ('CodeSystem', '$lookup'): [ - SYSTEM, VERSION, CODE, coding('code and system', ' version still selects the code system version.'), + LOOKUP_SYSTEM, VERSION, CODE, + coding('code and system', ' version still selects the code system version; at an instance, a separate ' + 'system must still name it.'), LANGUAGE], ('CodeSystem', '$validate-code'): [ URL, VERSION, CODE, diff --git a/core/fhir/tests_swagger.py b/core/fhir/tests_swagger.py index fb59820da..bd8cd2e52 100644 --- a/core/fhir/tests_swagger.py +++ b/core/fhir/tests_swagger.py @@ -119,21 +119,27 @@ def test_operation_parameters_and_bodies(self): if bodies: self.assertEqual(operation['consumes'], ['application/fhir+json']) self.assertEqual(bodies[0]['schema']['$ref'], '#/definitions/FhirParameters') - # a coding replaces these inputs (core.fhir.ops: coded_input, translate.read_request); each says so + expand = self.schema['paths']['/fhir/ValueSet/$expand']['get']['parameters'] + count = next(p for p in expand if p['name'] == 'count') + self.assertEqual((count['type'], count['maximum']), ('integer', 1000)) + + def test_coding_says_which_inputs_it_replaces(self): + # a coding replaces these inputs (core.fhir.ops: coded_input, translate.read_request), and the coding's + # description names them; those it leaves unread say so ($lookup's system still names an instance: mismatch) replaced = { - 'CodeSystem/$lookup': {'code', 'system'}, - 'CodeSystem/$validate-code': {'code', 'display'}, - 'ValueSet/$validate-code': {'code', 'system', 'systemVersion', 'display'}, - 'ConceptMap/$translate': {'code', 'system', 'version'}, + 'CodeSystem/$lookup': ({'code', 'system'}, {'code'}), + 'CodeSystem/$validate-code': ({'code', 'display'}, {'code', 'display'}), + 'ValueSet/$validate-code': ({'code', 'system', 'systemVersion', 'display'}, + {'code', 'system', 'systemVersion', 'display'}), + 'ConceptMap/$translate': ({'code', 'system', 'version'}, {'code', 'system', 'version'}), } - for path, names in replaced.items(): + for path, (names, unread) in replaced.items(): query = {p['name']: p for p in self.schema['paths'][f'/fhir/{path}']['get']['parameters']} for name in names: self.assertIn(name, query['coding']['description'], (path, name)) - self.assertIn('Not read when coding', query[name]['description'], (path, name)) - expand = self.schema['paths']['/fhir/ValueSet/$expand']['get']['parameters'] - count = next(p for p in expand if p['name'] == 'count') - self.assertEqual((count['type'], count['maximum']), ('integer', 1000)) + self.assertEqual('Not read when coding' in query[name]['description'], name in unread, (path, name)) + lookup = {p['name']: p for p in self.schema['paths']['/fhir/CodeSystem/$lookup']['get']['parameters']} + self.assertIn('At an instance it must name that code system', lookup['system']['description']) def test_tags_order_responses_and_security_definitions(self): for tag in (*OPERATIONS, 'Capabilities'):