135 lines
4.7 KiB
Python
135 lines
4.7 KiB
Python
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)
|
|
)
|