1fa005eb8a
Tracking docs/openapi.json (so the app can serve the spec at runtime) makes docs/ exist on the published GitHub mirror, which defeated the skipif(not DOCS.is_dir()) guard in the docs publish/version tests: they then ran against a prose-less checkout and failed on CI. Key the skip off docs/*.md prose instead, matching the canonical published-checkout definition in scripts/check-gates-ran.sh. Also: the OpenAPI generator emitted a trailing-slash path (/api/docs/), which a strict proxy like IIS treats as a redirecting resource a subpath-mounted client cannot follow; normalise trailing slashes in _norm_path and correct the committed spec. And allow docx (python-docx) as an undeclared import: it is used only by scripts/docs/generate_docx.py, a developer tool that never runs on a deployed site.
324 lines
13 KiB
Python
324 lines
13 KiB
Python
"""Generate docs/openapi.json from docs/api-inventory.json.
|
|
|
|
The inventory is a list of {surface, endpoints:[{method,path,auth,params,purpose,
|
|
example}]} objects (one per API surface). It is HAND-MAINTAINED - nothing
|
|
generates it, and the docstring here used to imply otherwise. That is deliberate:
|
|
its value is the prose, and what an endpoint is FOR cannot be derived from the
|
|
code. Coverage is enforced instead, by tests/test_api_inventory_parity.py, which
|
|
fails when a served route has no entry.
|
|
|
|
So: describe the endpoint in api-inventory.json, then re-run this. Served
|
|
interactively at /api/docs (see shopdb/core/api/docs.py).
|
|
|
|
venv/bin/python scripts/gen_openapi.py
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO = os.path.dirname(HERE)
|
|
INVENTORY = os.path.join(REPO, 'docs', 'api-inventory.json')
|
|
OUT = os.path.join(REPO, 'docs', 'openapi.json')
|
|
|
|
VERBS = ('get', 'post', 'put', 'patch', 'delete', 'head', 'options')
|
|
|
|
|
|
def _servers():
|
|
"""Servers block for the spec: the relative mount, plus this site's own URL
|
|
if one was supplied.
|
|
|
|
SHOPDB_PUBLIC_URL is read from the environment rather than stored, because
|
|
the generated spec is published to a public wiki - a site URL baked into the
|
|
generator ends up in everyone's documentation, including sites it is wrong
|
|
for.
|
|
"""
|
|
servers = []
|
|
siteurl = (os.environ.get('SHOPDB_PUBLIC_URL') or '').strip().rstrip('/')
|
|
if siteurl:
|
|
servers.append({'url': siteurl, 'description': 'this site'})
|
|
servers.append({'url': '/', 'description': 'relative to the deployed mount'})
|
|
return servers
|
|
|
|
|
|
def _product_version():
|
|
"""Read __version__ out of shopdb/__init__.py without importing the app.
|
|
|
|
Importing shopdb here would pull in Flask, the plugin loader and a database
|
|
configuration, none of which this generator needs.
|
|
"""
|
|
here = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
init = os.path.join(here, 'shopdb', '__init__.py')
|
|
with open(init, 'r', encoding='utf-8') as handle:
|
|
match = re.search(r"^__version__ = '([^']+)'", handle.read(), re.M)
|
|
if not match:
|
|
raise SystemExit('could not read __version__ from ' + init)
|
|
return match.group(1)
|
|
|
|
|
|
def _norm_path(p):
|
|
# Flask <int:id> / <id> -> OpenAPI {id}
|
|
normalised = re.sub(r'<(?:[^:>]+:)?([^>]+)>', r'{\1}', p or '')
|
|
# A trailing slash in a published path is a distinct resource to a strict
|
|
# proxy (IIS), which answers the unslashed URL with a redirect a client
|
|
# behind a subpath mount cannot follow. Emit the canonical, slashless form.
|
|
if len(normalised) > 1:
|
|
normalised = normalised.rstrip('/')
|
|
return normalised
|
|
|
|
|
|
def _security(auth):
|
|
"""OpenAPI security for one endpoint's documented auth string.
|
|
|
|
`jwt-optional` is the interesting case, and the common one - 121 endpoints.
|
|
It means "works logged out, returns more when logged in", which OpenAPI
|
|
expresses as a list containing BOTH the empty requirement and the scheme.
|
|
Publishing those as bearer-required told every reader, human and machine,
|
|
that a public endpoint needs a token.
|
|
"""
|
|
a = (auth or '').lower()
|
|
if 'optional' in a:
|
|
return [{}, {'bearerAuth': []}]
|
|
if a in ('none', 'public', '') or a.startswith('none') or a.startswith('public'):
|
|
return []
|
|
if 'api-key' in a or 'api_key' in a or 'x-api-key' in a:
|
|
return [{'apiKeyAuth': []}]
|
|
return [{'bearerAuth': []}]
|
|
|
|
|
|
def _path_parameters(rawpath):
|
|
"""Path parameters, typed from the Flask converter that declared them.
|
|
|
|
The generator emitted no `parameters` at all, which makes the document
|
|
invalid OpenAPI 3.1 - and breaks its only machine consumer outright, because
|
|
a tool built from an operation with no parameters has nowhere to put the id
|
|
the caller supplied, so it is dropped in silence.
|
|
"""
|
|
parameters = []
|
|
for converter, name in re.findall(r'<(?:([^:>]+):)?([^>]+)>', rawpath or ''):
|
|
schema = {'type': 'integer'} if converter in ('int',) else {'type': 'string'}
|
|
if converter == 'float':
|
|
schema = {'type': 'number'}
|
|
if converter == 'path':
|
|
schema = {'type': 'string', 'description': 'may contain slashes'}
|
|
parameters.append({
|
|
'name': name,
|
|
'in': 'path',
|
|
'required': True,
|
|
'schema': schema,
|
|
})
|
|
return parameters
|
|
|
|
|
|
def _request_body(verb, entry):
|
|
"""A JSON body for the verbs that take one.
|
|
|
|
The inventory describes bodies in prose, not as schemas, so this does not
|
|
invent field names it cannot verify - it declares an object and carries the
|
|
prose. That is honest, and it is the difference between a machine client
|
|
that can send a body and one that cannot send anything at all.
|
|
"""
|
|
if verb not in ('post', 'put', 'patch'):
|
|
return None
|
|
described = entry.get('params') or ''
|
|
return {
|
|
'required': verb in ('post', 'put'),
|
|
'content': {
|
|
'application/json': {
|
|
'schema': {
|
|
'type': 'object',
|
|
'additionalProperties': True,
|
|
'description': described or 'See the endpoint description.',
|
|
}
|
|
}
|
|
},
|
|
}
|
|
|
|
|
|
def _responses(security, parameters):
|
|
"""The envelope, plus the failures a caller has to handle.
|
|
|
|
One hardcoded 200 was the whole response section, so a generated client had
|
|
no idea any call could fail, and 19 operations that do not return the
|
|
envelope at all were documented as if they did.
|
|
"""
|
|
responses = {
|
|
'200': {
|
|
'description': 'Success. Body is the success_response envelope: '
|
|
'{status, data, meta}.',
|
|
'content': {'application/json': {
|
|
'$ref': '#/components/schemas/SuccessEnvelope'}},
|
|
},
|
|
'default': {
|
|
'description': 'Error. Body is the error envelope; the code and '
|
|
'message are nested under data.error.',
|
|
'content': {'application/json': {
|
|
'$ref': '#/components/schemas/ErrorEnvelope'}},
|
|
},
|
|
}
|
|
# An endpoint that can be called without a token cannot answer 401.
|
|
if security and security != [{}]:
|
|
requires = not any(item == {} for item in security)
|
|
if requires:
|
|
responses['401'] = {'description': 'Missing or invalid credentials.'}
|
|
responses['403'] = {'description': 'Authenticated, but not permitted.'}
|
|
if parameters:
|
|
responses['404'] = {'description': 'No such record.'}
|
|
return responses
|
|
|
|
|
|
def _summary(text, fallback):
|
|
"""A summary that ends on a word.
|
|
|
|
Cutting at a fixed 120 characters truncated 95 of them mid-word, which is
|
|
what a tool picker shows a user as the whole description of the call.
|
|
"""
|
|
source = (text or fallback or '').strip()
|
|
if len(source) <= 120:
|
|
return source
|
|
head = source[:120]
|
|
cut = head.rfind(' ')
|
|
# Trailing connectives read worse than a clean cut: "servicelevel +..."
|
|
# promises a continuation the reader will never see.
|
|
return (head[:cut] if cut > 40 else head).rstrip(' ,;:.+-/&|(') + '...'
|
|
|
|
|
|
def build(surfaces):
|
|
paths = {}
|
|
tags = []
|
|
for s in surfaces:
|
|
tags.append({'name': s['surface']})
|
|
for e in s.get('endpoints', []):
|
|
# 'PUT|PATCH' is how the inventory writes a multi-verb route.
|
|
# Without the pipe here, neither verb matched VERBS and the
|
|
# operation vanished from the spec without a word.
|
|
for verb in re.split(r'[\/,|]', (e.get('method') or 'GET')):
|
|
verb = verb.strip().lower()
|
|
if verb not in VERBS:
|
|
continue
|
|
path = _norm_path(e.get('path'))
|
|
if not path:
|
|
continue
|
|
desc = []
|
|
if e.get('purpose'):
|
|
desc.append(e['purpose'])
|
|
if e.get('auth'):
|
|
desc.append('\n\n**Auth:** ' + e['auth'])
|
|
if e.get('params'):
|
|
desc.append('\n\n**Params:** ' + e['params'])
|
|
if e.get('example'):
|
|
desc.append('\n\n**Example:**\n```\n' + e['example'] + '\n```')
|
|
security = _security(e.get('auth'))
|
|
parameters = _path_parameters(e.get('path'))
|
|
operation = {
|
|
'tags': [s['surface']],
|
|
'summary': _summary(e.get('purpose'), path),
|
|
'description': ''.join(desc),
|
|
'security': security,
|
|
'responses': _responses(security, parameters),
|
|
}
|
|
if parameters:
|
|
operation['parameters'] = parameters
|
|
body = _request_body(verb, e)
|
|
if body:
|
|
operation['requestBody'] = body
|
|
paths.setdefault(path, {})[verb] = operation
|
|
return {
|
|
'openapi': '3.1.0',
|
|
'info': {
|
|
'title': 'ShopDB Flask API',
|
|
# Read from the code, not restated. A hardcoded copy here had
|
|
# already drifted a release behind, and the same mistake in the
|
|
# installer script shipped an exe stamped with the wrong version.
|
|
'version': _product_version(),
|
|
'description': (
|
|
'Asset-management API (core + plugins). Responses use a '
|
|
'`success_response` envelope: `{status, data, meta}`. Auth: Bearer '
|
|
'JWT (login or a managed PAT) for `jwt`/`admin`/`permission:*`; '
|
|
'`X-API-Key` for collector/managed-token endpoints; public endpoints '
|
|
'need neither.'),
|
|
},
|
|
# One site's production hostname was hardcoded here, so every generated
|
|
# spec published it to the public wiki and offered a second site a
|
|
# server it cannot reach. The relative mount is the only server this
|
|
# generator can honestly name; a site that wants its own in the spec
|
|
# sets SHOPDB_PUBLIC_URL when regenerating.
|
|
'servers': _servers(),
|
|
'components': {
|
|
'schemas': {
|
|
# The envelope every JSON endpoint returns. Worth spelling out
|
|
# because the error shape nests one level deeper than most
|
|
# people assume, and code written against the assumption reads
|
|
# undefined on every error it tries to report.
|
|
'SuccessEnvelope': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'status': {'type': 'string', 'enum': ['success']},
|
|
'data': {'description': 'The payload. Shape is per endpoint.'},
|
|
'meta': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'timestamp': {'type': 'string', 'format': 'date-time'},
|
|
'requestid': {'type': 'string'},
|
|
'pagination': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'page': {'type': 'integer'},
|
|
'perpage': {'type': 'integer'},
|
|
'total': {'type': 'integer'},
|
|
'pages': {'type': 'integer'},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
},
|
|
'required': ['status'],
|
|
},
|
|
'ErrorEnvelope': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'status': {'type': 'string', 'enum': ['error']},
|
|
'data': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'error': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'code': {'type': 'string'},
|
|
'message': {'type': 'string'},
|
|
'details': {'type': 'object'},
|
|
},
|
|
'required': ['code', 'message'],
|
|
},
|
|
},
|
|
'required': ['error'],
|
|
},
|
|
'meta': {'type': 'object'},
|
|
},
|
|
'required': ['status', 'data'],
|
|
},
|
|
},
|
|
'securitySchemes': {
|
|
'bearerAuth': {'type': 'http', 'scheme': 'bearer', 'bearerFormat': 'JWT',
|
|
'description': 'Login token or a managed Personal Access Token (scoped).'},
|
|
'apiKeyAuth': {'type': 'apiKey', 'in': 'header', 'name': 'X-API-Key',
|
|
'description': 'Managed service token (e.g. collector.ingest, geenforce.fetch).'},
|
|
}},
|
|
'tags': tags,
|
|
'paths': paths,
|
|
}
|
|
|
|
|
|
def main():
|
|
surfaces = json.load(open(INVENTORY))
|
|
spec = build(surfaces)
|
|
json.dump(spec, open(OUT, 'w'), indent=1)
|
|
ops = sum(len(v) for v in spec['paths'].values())
|
|
print('wrote %s: %d paths, %d operations from %d surfaces'
|
|
% (OUT, len(spec['paths']), ops, len(surfaces)))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|