-
Notifications
You must be signed in to change notification settings - Fork 6
/
Copy path__init__.py
59 lines (46 loc) · 1.73 KB
/
__init__.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
from typing import List, Optional
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from sqlalchemy import select
from sqlalchemy.orm import Session, subqueryload
from .config import settings
from .database import SessionLocal
from .enum import FilterEnum
from .dynamic_routes import create_dynamic_router
from .models import Schema, AttributeDefinition, AttrType
from .general_routes import router
def load_schemas(db: Session) -> List[models.Schema]:
schemas = db.execute(
select(Schema)
.options(
subqueryload(Schema.attr_defs)
)
).scalars().all()
return schemas
def generate_api_description() -> str:
description = '# Filters\n\n**Filters list**:'
for filter in FilterEnum:
description += '\n* `{}` - {}'.format(filter.value.name, filter.value.description)
description += '\n\n**Available filters for each type**:'
for atype in AttrType:
description += '\n* `{}`: {}'.format(atype.name, ', '.join([f'`{i.value.name}`' for i in atype.value.filters]))
return description
def load_dynamic_routes(db: Session, app: FastAPI):
schemas = load_schemas(db)
for schema in schemas:
create_dynamic_router(schema=schema, app=app)
def create_app(session: Optional[Session] = None) -> FastAPI:
app = FastAPI(description=generate_api_description())
origins = ['*']
app.add_middleware(CORSMiddleware,
allow_origins=origins,
allow_credentials=True,
allow_methods=['*'],
allow_headers=['*'])
if session:
load_dynamic_routes(db=session, app=app)
else:
with SessionLocal() as db:
load_dynamic_routes(db=db, app=app)
app.include_router(router)
return app