1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89
|
import json
import tempfile
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.openapi.utils import get_openapi
from mangum import Mangum
from stac_validator import stac_validator
# app = FastAPI() # OpenAPI docs are in a custom function below.
# this is used to push to aws cdk with prod endpoint
app = FastAPI(title="STAC Validator", version=2.0, root_path="/prod/")
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
def validate(stac_object):
stac = stac_validator.StacValidate(stac_object)
stac.run()
output = stac.message[0]
return output
@app.get("/")
async def homepage():
return {"body": "https://api.staclint.com/docs"}
@app.get("/url")
async def validate_url_get_request(stac_url):
output = validate(str(stac_url))
return {"body": output}
@app.post("/url")
async def validate_url_post_request(stac_url):
output = validate(str(stac_url))
return {"body": output}
@app.post("/json")
async def validate_json(request: Request):
stac_file = await request.json()
temp_stac_file = tempfile.NamedTemporaryFile(
delete=False,
mode="w+",
)
json.dump(stac_file, temp_stac_file)
temp_stac_file.flush()
temp_stac_file.close()
body = temp_stac_file.name
output = validate(body)
return {"body": output}
def custom_openapi():
if app.openapi_schema:
return app.openapi_schema
openapi_schema = get_openapi(
title="STAC Validator",
description="API for validating STAC files. Powered by Sparkgeo.",
version=2.0,
routes=app.routes,
)
# openapi_schema["paths"]["/url"]["get"] = {
# "description": "This endpoint supports validation of STAC file. Use stac_url as the query parameter.",
# }
# openapi_schema["paths"]["/url"]["post"] = {
# "description": "This endpoint supports validation of a STAC file. Post your data as JSON with stac_url as the key and your path as the value.",
# }
openapi_schema["paths"]["/json"]["post"] = {
"description": "This endpoint supports validation of STAC JSON directly. Post your data as JSON.",
}
app.openapi_schema = openapi_schema
return app.openapi_schema
app.openapi = custom_openapi
handler = Mangum(app)
|