67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
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
|