Verified Commit a00c80f2 authored by Dan Jones's avatar Dan Jones
Browse files

refactor: load geojson defs from published schemas

parent a0d9c041
No related merge requests found
Pipeline #227529 failed with stages
in 18 seconds
......@@ -184,20 +184,11 @@ platform_schema = {
region_schema = {
"type": "object",
"properties": {
"geometry_coordinates": {
"type": "array",
"example": [
[
[-4.1777839187560915, 50.34173405662855],
[-4.1777839187560915, 50.33820949229701],
[-4.143667777943875, 50.33820949229701],
[-4.143667777943875, 50.34173405662855],
[-4.1777839187560915, 50.34173405662855],
]
],
"geometry": {
"$ref": "https://geojson.org/schema/Polygon.json",
},
},
"description": "Using GEOJSON, exact 4-point region (rectangle shaped - 5 points)",
"description": "GeoJSON Polygon",
}
squad_metadata_schema = {
......
......@@ -14,9 +14,11 @@ from formats.alert import alert_schema
from flasgger import Swagger
from flask import Flask
import json
import os
import re
import requests
app = Flask(__name__)
url_prefix = os.getenv("URL_PREFIX", "")
......@@ -103,7 +105,98 @@ swagger_config = {
},
}
swag = Swagger(app, config=swagger_config, merge=True)
def resolve_ref(ref):
"""
Get schema URL, parse JSON
Return None if either fails
"""
try:
res = requests.get(ref)
if res.status_code == 200:
return res.json()
else:
return None
except (json.JSONDecodeError, ValueError):
return None
def rename_ref(ref):
"""
Convert remote ref URL into a name that can
be used for a local ref in the schema
Remote the URL scheme and replace / with .
"""
# remove url scheme
deschemed = re.sub(r"^[htps]*\:*[/]{2}", "", ref)
# replace / with . since the name will be in a path
return re.sub(r"[/]", ".", deschemed)
def nested_replace(source, key, value, replace_with):
"""
Find all instances of a key value pair in a nested
dictionary and replace the value with replace_with
"""
for k,v in source.items():
if k == key and v == value:
source[k] = replace_with
elif type(v) is list:
for item in v:
if type(item) is dict:
nested_replace(item, key, value, replace_with)
if type(v) is dict:
nested_replace(v, key, value, replace_with)
def inject_schema(schema, remote_ref):
"""
Given a parent schema and a remote ref
1. get the remote ref schema
2. create a local reference name (without path separators)
3. insert into components.schemas
4. replace remote references with local references
returns True if resolved and injected
"""
local_name = rename_ref(remote_ref)
local_ref = f"#/components/schemas/{local_name}"
ref_schema = resolve_ref(remote_ref)
if (ref_schema is not None):
nested_replace(schema, "$ref", remote_ref, local_ref)
schema["components"]["schemas"][local_name] = ref_schema
return True
else:
return False
def import_remote_refs():
"""
inject the following remote refs into the schema
and replace the remote refs with local refs
returns True if all schemas resolved and injected
"""
ref_imports = [
"https://geojson.org/schema/Feature.json",
"https://geojson.org/schema/FeatureCollection.json",
"https://geojson.org/schema/LineString.json",
"https://geojson.org/schema/Point.json",
"https://geojson.org/schema/Polygon.json",
]
return all([
inject_schema(swagger_config, ref)
for ref in ref_imports
])
import_remote_refs()
app = Flask(__name__)
Swagger(app, config=swagger_config, merge=True)
flask_host = os.getenv(
"FLASK_HOST", "localhost"
......
This diff is collapsed.
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment