from fastapi import FastAPI from pydantic import BaseModel from typing import List from sqlalchemy import create_engine, text import psycopg2.extras from psycopg2.extensions import AsIs, register_adapter from dataclasses import dataclass from sqlalchemy.dialects.postgresql import JSONB app = FastAPI() engine = create_engine("postgresql://postgres?host=/home/anish/usr/df-web/postgres/") conn = engine.connect() psycopg2.extensions.register_adapter(dict, psycopg2.extras.Json) class vector3(BaseModel): x: float y: float z: float class Route(BaseModel): id: str creatorName: str dustPoints: list # List[Vector3] checkpoints: list # List[Vector3] startLocation: dict # Vector3 endLocation: dict # Vector3 creationDate: str @app.get("/route") async def get_route(): select = "SELECT * FROM routes" result = engine.execute(select) data = [] for route in result: data.append(route) return {"data": data} @app.post("/route") async def post_route(route: Route): as_dict = route.dict() insert = text( """INSERT INTO routes (id, creatorName, creationDate, startLocation, endLocation, checkpoints, dustPoints) VALUES (:id, :creatorName, creationDate CAST(:startLocation as jsonb), CAST(:endLocation as jsonb), CAST(:checkpoints as jsonb[]), CAST(:dustPoints as jsonb[]))""" ).bindparams(**as_dict) engine.execute(insert) return