File: lambda.py

package info (click to toggle)
stac-validator 3.6.0%2Bds-1
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 2,044 kB
  • sloc: python: 2,868; makefile: 91
file content (89 lines) | stat: -rw-r--r-- 2,395 bytes parent folder | download
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)