service.py
1 import falcon 2 import psycopg2 3 import psycopg2.extras 4 from datetime import datetime, date 5 from db_utils import get_db_connection, json_serial 6 7 class HelloDBResource: 8 def on_get(self, req, resp): 9 _test_mode = req.get_header('X-Test-Request') == 'true' 10 try: 11 with get_db_connection(test_mode=_test_mode) as conn: 12 with conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: 13 cur.execute("SELECT * FROM helloDB") 14 results = cur.fetchall() 15 16 # Serialize datetime columns to strings 17 serialized_results = [ 18 { 19 k: json_serial(v) if isinstance(v, (datetime, date)) else v 20 for k, v in row.items() 21 } 22 for row in results 23 ] 24 25 resp.media = { 26 "message": "Hello, Database!", 27 "status": "success", 28 "data": serialized_results, 29 } 30 resp.status = falcon.HTTP_200 31 32 except psycopg2.Error as e: 33 resp.media = {"message": f"Database error: {str(e)}", "status": "error"} 34 resp.status = falcon.HTTP_500 35 36 except Exception as e: 37 resp.media = {"message": f"Server error: {str(e)}", "status": "error"} 38 resp.status = falcon.HTTP_500 39 40 class HelloDBSchemaResource: 41 def on_get(self, req, resp): 42 return [ 43 {"field": "id", "headerName": "ID", "type": "number", "width": 70}, 44 {"field": "name", "headerName": "Name", "type": "string", "width": 150}, 45 {"field": "age", "headerName": "Age", "type": "number", "width": 100}, 46 {"field": "email", "headerName": "Email", "type": "string", "width": 200}, 47 {"field": "gender", "headerName": "Gender", "type": "enum", "enumValues":["Male","Female"], "width": 200}, 48 {"field": "companyId", "headerName": "Company Id", "type": "foreignKey", "relatedTable":"company", "relatedColumn":"id", "width": 200}, 49 ] 50