df-web/main.py

from fastapi import FastAPI
from pydantic import BaseModel
from http import HTTPStatus
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):
    route: dict
    # 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


@app.post("/route")
async def post_route(route: dict):
    print(route)
    # as_dict = route.dict()
    insert = text(
        """INSERT INTO
                  routes (id, "creatorName", "creationDate", name, "creatorCharacter", "startLocation", "endLocation", checkpoints, dustpoints)
                  VALUES
                  (:id, :creatorName, :creationDate, :name, :creatorCharacter,
                  CAST(:startLocation as jsonb),
                  CAST(:endLocation as jsonb),
                  CAST(:checkpoints as jsonb[]),
                  CAST(:dustpoints as jsonb[]))"""
    ).bindparams(**route)
    engine.execute(insert)
    return HTTPStatus.OK


@app.get("/route/delete")
async def delete_routes():
    delete = text("TRUNCATE TABLE routes")
    engine.execute(delete)
    return HTTPStatus.OK