import logging import uvicorn from arel import HotReload, Path from database import engine, Base from dotenv import load_dotenv from fastapi import FastAPI, Request, status, Header from fastapi.encoders import jsonable_encoder from fastapi.responses import JSONResponse, HTMLResponse from fastapi.staticfiles import StaticFiles from fastapi.templating import Jinja2Templates from loggerino import timestamp_log_config from os import getenv from schemas import SMSMessage from service import ( receive_sms_messages_handler, retrieve_sms_messages_by_phone_number_handler, delete_sms_message_by_uuid_handler ) from typing import Optional from uuid import UUID from uvicorn.config import LOGGING_CONFIG # -------------------------------------------------------------------------------- # Bootstrap # -------------------------------------------------------------------------------- load_dotenv() # -------------------------------------------------------------------------------- # App Creation # -------------------------------------------------------------------------------- app = FastAPI() # -------------------------------------------------------------------------------- # Logger # -------------------------------------------------------------------------------- logger = logging.getLogger("uvicorn.error") # -------------------------------------------------------------------------------- # Database # -------------------------------------------------------------------------------- Base.metadata.create_all(bind=engine) # -------------------------------------------------------------------------------- # Application Configurations # -------------------------------------------------------------------------------- WHITELISTED_IPS = getenv("WHITELISTED_IPS").split(',') # -------------------------------------------------------------------------------- # Static Files # -------------------------------------------------------------------------------- app.mount("/static", StaticFiles(directory="static"), name="static") # -------------------------------------------------------------------------------- # Templates # -------------------------------------------------------------------------------- templates = Jinja2Templates(directory="templates") # -------------------------------------------------------------------------------- # Hot Reload # -------------------------------------------------------------------------------- if _debug := getenv("DEBUG"): hot_reload = HotReload(paths=[Path(".")]) app.add_websocket_route("/hot-reload", route=hot_reload, name="hot-reload") app.add_event_handler("startup", hot_reload.startup) app.add_event_handler("shutdown", hot_reload.shutdown) templates.env.globals["DEBUG"] = _debug templates.env.globals["hot_reload"] = hot_reload # -------------------------------------------------------------------------------- # Middleware # -------------------------------------------------------------------------------- @app.middleware("http") async def validate_ip(request: Request, call_next): ip = str(request.headers.get("x-forwarded-for", str(request.client.host))) if ip not in WHITELISTED_IPS: data = {"message": f"IP {ip} is not allowed to access this resource."} return JSONResponse(status_code=status.HTTP_400_BAD_REQUEST, content=data) return await call_next(request) # -------------------------------------------------------------------------------- # Routes # -------------------------------------------------------------------------------- @app.get("/", summary="Main page that redirects to application", response_class=HTMLResponse) async def index(request: Request): return templates.TemplateResponse("index.html", {"request": request}) @app.get('/sms-message/{number}', response_class=HTMLResponse) def get_sms_messages_by_phone_number(request: Request, number: str, hx_request: Optional[str] = Header(None), cost: bool = False, metadata: bool = False): contact = retrieve_sms_messages_by_phone_number_handler(number, cost, metadata) if hx_request: return templates.TemplateResponse( request=request, name="messages.html", context={"messages": contact} ) return JSONResponse(content=jsonable_encoder({'status': 'success', 'results': len(contact), 'response': contact})) @app.post("/sms-message", status_code=status.HTTP_200_OK) async def post_sms_message(message: SMSMessage): receive_sms_messages_handler(message) @app.delete('/sms-message/{uuid}', status_code=status.HTTP_202_ACCEPTED) def delete_sms_message_by_uuid(uuid: UUID): delete_sms_message_by_uuid_handler(uuid) if __name__ == "__main__": uvicorn.run( app, host="0.0.0.0", port=8000, log_config=timestamp_log_config(LOGGING_CONFIG) )