-
Notifications
You must be signed in to change notification settings - Fork 8
/
main.py
75 lines (59 loc) · 2.25 KB
/
main.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
import os
from contextlib import asynccontextmanager
from fastapi import Depends, FastAPI
from scalar_fastapi import get_scalar_api_reference
from sqlalchemy import inspect
from sqlalchemy.orm import Session
from shared.constants import PROJECT_VERSION
from shared.log_config import get_logger
from trustregistry import crud, db
from trustregistry.database import engine
from trustregistry.db import get_db
from trustregistry.registry import registry_actors, registry_schemas
logger = get_logger(__name__)
OPENAPI_NAME = os.getenv("OPENAPI_NAME", "Trust Registry")
ROOT_PATH = os.getenv("ROOT_PATH", "")
@asynccontextmanager
async def lifespan(_: FastAPI):
db.Base.metadata.create_all(bind=engine)
engine.dispose()
logger.debug("TrustRegistry startup: Validate tables are created")
with engine.connect() as connection:
inspector = inspect(connection)
table_names = inspector.get_table_names()
logger.debug("TrustRegistry tables created: `{}`", table_names)
# start-up logic is before the yield
yield
# shutdown logic after
def create_app():
application = FastAPI(
root_path=ROOT_PATH,
title=OPENAPI_NAME,
version=PROJECT_VERSION,
description="Welcome to the OpenAPI interface to the Aries CloudAPI trust registry",
lifespan=lifespan,
redoc_url=None,
docs_url=None,
)
application.include_router(registry_actors.router)
application.include_router(registry_schemas.router)
return application
app = create_app()
# Use Scalar instead of Swagger
@app.get("/docs", include_in_schema=False)
async def scalar_html():
return get_scalar_api_reference(
openapi_url=app.openapi_url,
title=app.title,
)
@app.get("/")
async def root(db_session: Session = Depends(get_db)):
logger.debug("GET request received: Fetch actors and schemas from registry")
db_schemas = crud.get_schemas(db_session)
db_actors = crud.get_actors(db_session)
schemas_repr = [schema.id for schema in db_schemas]
logger.debug("Successfully fetched actors and schemas from registry.")
return {"actors": db_actors, "schemas": schemas_repr}
@app.get("/registry")
async def registry(db_session: Session = Depends(get_db)):
return await root(db_session)