/
api.py
222 lines (169 loc) · 6.21 KB
/
api.py
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
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
# -*- coding: utf-8 -*-
"""API"""
from typing import List, Optional, Union
from fastapi import FastAPI, HTTPException, Query
from pydantic import BaseModel
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import RedirectResponse
from . import __description__, __projectname__, __version__
from .declaration import (
AS3Declaration,
AS3JSONDecodeError,
AS3TemplateSyntaxError,
AS3UndefinedError,
)
from .gitget import Gitget, GitgetException
from .schema import AS3Schema, AS3SchemaVersionError, AS3ValidationError
from .templateconfiguration import AS3TemplateConfiguration, AS3TemplateConfigurationError
from .utils import deserialize
CORS_SETTINGS = {
"allow_origins": [
"http://localhost",
"http://localhost:8000",
"https://localhost",
"https://localhost:8000",
],
"allow_credentials": True,
"allow_methods": ["*"],
"allow_headers": ["*"],
}
class AS3ValidationResult(BaseModel):
"""AS3 declaration Schema validation result"""
valid: bool
error: Optional[str]
class LatestVersion(BaseModel):
"""AS3 /schema/latest_version response"""
latest_version: str
class Error(BaseModel):
"""Generic Error Model"""
code: int
message: str
class AS3DeclareGit(BaseModel):
"""Model for an AS3 Declaration from a Git repository"""
repository: str
branch: Optional[str]
commit: Optional[str]
depth: int = 1
template_configuration: Optional[Union[List[Union[dict, str]], dict, str]]
class AS3Declare(BaseModel):
"""Model for an inline AS3 Declaration"""
template_configuration: Union[List[dict], dict]
declaration_template: str
app = FastAPI(docs_url=None, redoc_url=None, openapi_url=None)
app.add_middleware(CORSMiddleware, **CORS_SETTINGS)
@app.on_event("startup")
def startup():
# preload AS3Schema Class - assume Schemas are available
_ = AS3Schema()
@app.get("/")
async def default_redirect():
"""redirect / to /api/docs"""
return RedirectResponse(url="/api/docs")
@app.get("/docs")
async def docs_redirect():
"""redirect /docs to /api/docs"""
return RedirectResponse(url="/api/docs")
@app.get("/redoc")
async def redoc_redirect():
"""redirect /redoc to /api/redoc"""
return RedirectResponse(url="/api/redoc")
@app.get("/openapi.json")
async def openapi_redirect():
"""redirect /openapi.json to /api/openapi.json"""
return RedirectResponse(url="/api/openapi.json")
api = FastAPI(
openapi_prefix="/api",
title=__projectname__,
description=__description__,
version=__version__,
)
api.add_middleware(CORSMiddleware, **CORS_SETTINGS)
@api.get("/schema/latest_version")
async def get_schema_latest_version():
"""Returns latest known AS3 Schema version"""
return LatestVersion(latest_version=AS3Schema().latest_version)
@api.get("/schema/schema")
async def get_schema_schema_version(
version: str = Query("latest", title="AS3 Schema version to get"),
):
"""Returns AS3 Schema of ``version``"""
try:
return AS3Schema(version=version).schema
except AS3SchemaVersionError as exc:
error = Error(code=400, message=str(exc))
raise HTTPException(status_code=error.code, detail=error.message)
@api.get("/schema/schemas")
async def get_schema_schemas():
"""Returns all known AS3 Schemas"""
return AS3Schema().schemas
@api.get("/schema/versions")
async def get_schema_versions():
"""Returns array of version numbers for all known AS3 Schemas"""
return AS3Schema().versions
@api.post("/schema/validate", response_model=AS3ValidationResult)
async def _schema_validate(
declaration: dict,
version: str = Query("latest", title="AS3 Schema version to validation against"),
):
"""Validate declaration in POST payload against AS3 Schema of ``version`` (Default: latest)"""
try:
AS3Schema(version=version).validate(declaration=declaration)
return AS3ValidationResult(valid=True)
except AS3SchemaVersionError as exc:
error = Error(code=400, message=str(exc))
raise HTTPException(status_code=400, detail=error.message)
except AS3ValidationError as exc:
return AS3ValidationResult(valid=False, error=str(exc))
@api.post("/declaration/transform")
async def post_declaration_transform(as3d: AS3Declare):
"""Transforms an AS3 declaration template, see ``AS3Declare`` for details on the expected input. Returns the AS3 Declaration."""
error = None
try:
as3tc = AS3TemplateConfiguration(as3d.template_configuration)
as3declaration = AS3Declaration(
template_configuration=as3tc.dict(),
declaration_template=as3d.declaration_template,
)
return as3declaration.dict()
except (
AS3SchemaVersionError,
AS3JSONDecodeError,
AS3TemplateSyntaxError,
AS3UndefinedError,
AS3TemplateConfigurationError,
) as exc:
error = Error(code=400, message=str(exc))
raise HTTPException(status_code=error.code, detail=error.message)
@api.post("/declaration/transform/git")
async def post_declaration_git_transform(as3d: AS3DeclareGit):
"""Transforms an AS3 declaration template, see ``AS3DeclareGit`` for details on the expected input. Returns the AS3 Declaration."""
error = None
try:
with Gitget(
repository=as3d.repository,
branch=as3d.branch,
commit=as3d.commit,
depth=as3d.depth,
) as gitrepo:
as3tc = AS3TemplateConfiguration(
template_configuration=as3d.template_configuration,
base_path=f"{gitrepo.repodir}/",
overlay={"as3ninja": {"git": gitrepo.info}},
)
as3declaration = AS3Declaration(
template_configuration=as3tc.dict(),
jinja2_searchpath=gitrepo.repodir,
)
return as3declaration.dict()
except (
GitgetException,
AS3SchemaVersionError,
AS3JSONDecodeError,
AS3TemplateSyntaxError,
AS3UndefinedError,
AS3TemplateConfigurationError,
) as exc:
error = Error(code=400, message=str(exc))
raise HTTPException(status_code=error.code, detail=error.message)
# mount api
app.mount("/api", api)