import datetime import time import json # import firebase_admin import uvicorn from fastapi import FastAPI, Depends, Body from fastapi.staticfiles import StaticFiles from fastapi.encoders import jsonable_encoder from pydantic import BaseModel from sqlalchemy.exc import DBAPIError from starlette.requests import Request from starlette.responses import JSONResponse from auth import * app = FastAPI( title="Smart lock", root_path="/users" ) @app.get("/docs", include_in_schema=False) async def custom_swagger_ui_html(req: Request): root_path = req.scope.get("root_path", "").rstrip("/") openapi_url = root_path + app.openapi_url return get_swagger_ui_html( openapi_url=openapi_url, title="API", ) app.include_router( fastapi_users.get_auth_router(auth_backend), prefix="/auth", tags=["Auth"], ) app.include_router( fastapi_users.get_register_router(UserRead, UserCreate), prefix="/auth", tags=["Auth"], ) # firebase # cred = credentials.Certificate("../serviceAccountKey.json") # try: # fire_app = firebase_admin.initialize_app(cred) # except ValueError: # pass class Message(BaseModel): msg: str @app.post("/msg") def debug_message_from_user(msg: Message = Body(), user: User = Depends(current_user_optional)): print(datetime.datetime.now(), end=' ') if user is not None: print(user.__dict__) print(msg.msg) @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.perf_counter() response = await call_next(request) process_time = time.perf_counter() - start_time print(f"Finished in {process_time:.3f}") response.headers["X-Process-Time"] = str(process_time) return response @app.exception_handler(DBAPIError) def authjwt_exception_handler(request: Request, exc: DBAPIError): return JSONResponse( status_code=500, content={"detail": jsonable_encoder(exc) | {"args": jsonable_encoder(exc.args)}} ) if __name__ == '__main__': uvicorn.run(app=app, host="0.0.0.0", port=8000)