Spaces:
Running
Running
File size: 3,020 Bytes
c4d1e86 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 | from fastapi import APIRouter, Form, File, UploadFile
from utils.common import CommonResponse
from utils.db import get_connection
router = APIRouter(
prefix="/testdb",
tags=["testdb"]
)
@router.get("/")
def test_home():
try:
return CommonResponse(success=True)
except Exception as e:
return CommonResponse(success=False, msg=str(e))
@router.get("/now")
def select_now():
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("select now()")
row = cur.fetchone()
data = {
"now": row[0].isoformat() if row else None
}
return CommonResponse(success=True, data=data)
except Exception as e:
return CommonResponse(success=False, msg=str(e))
@router.get("/select_test")
def select_now(id: int = 0):
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT
*
FROM t_test
WHERE
CASE
WHEN %s != 0 THEN id = %s
ELSE TRUE
END
LIMIT 1000
""", [id, id])
row = cur.fetchall()
data = row
return CommonResponse(success=True, data=data)
except Exception as e:
return CommonResponse(success=False, msg=str(e))
@router.post("/upsert_test")
def upsert_test(id: int = Form(0), name: str = Form("")):
try:
with get_connection() as conn:
with conn.cursor() as cur:
if id <= 0:
cur.execute("""
INSERT INTO t_test (name)
VALUES (%s)
RETURNING id, name
""", [name])
else:
cur.execute("""
UPDATE t_test
SET name = %s
WHERE id = %s
RETURNING id, name
""", [name, id])
row = cur.fetchone()
if not row:
conn.rollback()
return CommonResponse(success=False, msg="data not found")
conn.commit()
data = {
"id": row[0],
"name": row[1]
}
return CommonResponse(success=True, data=data)
except Exception as e:
return CommonResponse(success=False, msg=str(e))
@router.post("/delete_test")
def delete_test(id: int = Form(0)):
try:
with get_connection() as conn:
with conn.cursor() as cur:
cur.execute("""
DELETE FROM t_test
WHERE id = %s
""", [id])
conn.commit()
data = {
"msg" : "delete success"
}
return CommonResponse(success=True, data=data)
except Exception as e:
return CommonResponse(success=False, msg=str(e))
|