1+ import os
2+ from jinja2 import Environment , FileSystemLoader
3+
4+ class CodeGenerator :
5+ def __init__ (self ):
6+ self .env = Environment (loader = FileSystemLoader ('templates' ))
7+ self .template_string = """
8+ import sqlite3
9+ from fastapi import FastAPI, HTTPException, Query, Depends
10+ from typing import List, Optional
11+ from pydantic import BaseModel
12+ from datetime import datetime
13+
14+ app = FastAPI()
15+
16+ DB_NAME = "{{ db_name }}"
17+
18+ def execute_query(query: str, params: tuple = ()) -> List[dict]:
19+ conn = sqlite3.connect(DB_NAME)
20+ conn.row_factory = sqlite3.Row
21+ cursor = conn.cursor()
22+ cursor.execute(query, params)
23+ result = [dict(row) for row in cursor.fetchall()]
24+ conn.commit()
25+ conn.close()
26+ return result
27+
28+ {% for table in tables %}
29+ class {{ table.name | capitalize }}Base(BaseModel):
30+ {% for column in table.columns %}{% if column.name != table.primary_key %}{{ column.name }}: {{ column.py_type }}{% if column.nullable %} = None{% endif %}
31+ {% endif %}{% endfor %}
32+
33+ class {{ table.name | capitalize }}Create({{ table.name | capitalize }}Base):
34+ pass
35+
36+ class {{ table.name | capitalize }}({{ table.name | capitalize }}Base):
37+ {{ table.primary_key }}: int
38+
39+ @app.get("/{{ table.name | lower }}", response_model=List[{{ table.name | capitalize }}])
40+ async def get_{{ table.name | lower }}(
41+ skip: int = Query(0, ge=0),
42+ limit: int = Query(100, ge=1, le=1000),
43+ sort_by: Optional[str] = None,
44+ order: Optional[str] = Query(None, pattern="^(asc|desc)$")
45+ ):
46+ query = "SELECT * FROM {{ table.name }}"
47+ if sort_by:
48+ query += f" ORDER BY {sort_by}"
49+ if order:
50+ query += f" {order.upper()}"
51+ query += " LIMIT ? OFFSET ?"
52+ return execute_query(query, (limit, skip))
53+
54+ @app.get("/{{ table.name | lower }}/{item_id}", response_model={{ table.name | capitalize }})
55+ async def get_{{ table.name | lower }}_item(item_id: int):
56+ query = "SELECT * FROM {{ table.name }} WHERE {{ table.primary_key }} = ?"
57+ result = execute_query(query, (item_id,))
58+ if not result:
59+ raise HTTPException(status_code=404, detail="Item not found")
60+ return result[0]
61+
62+ @app.post("/{{ table.name | lower }}", response_model={{ table.name | capitalize }})
63+ async def create_{{ table.name | lower }}(item: {{ table.name | capitalize }}Create):
64+ columns = ", ".join(item.dict().keys())
65+ placeholders = ", ".join("?" * len(item.dict()))
66+ query = f"INSERT INTO {{ table.name }} ({columns}) VALUES ({placeholders})"
67+ execute_query(query, tuple(item.dict().values()))
68+ return {**item.dict(), "{{ table.primary_key }}": execute_query("SELECT last_insert_rowid()")[0]['last_insert_rowid()']}
69+
70+ @app.put("/{{ table.name | lower }}/{item_id}", response_model={{ table.name | capitalize }})
71+ async def update_{{ table.name | lower }}(item_id: int, item: {{ table.name | capitalize }}Create):
72+ set_clause = ", ".join(f"{k} = ?" for k in item.dict().keys())
73+ query = f"UPDATE {{ table.name }} SET {set_clause} WHERE {{ table.primary_key }} = ?"
74+ values = tuple(item.dict().values()) + (item_id,)
75+ execute_query(query, values)
76+ return {**item.dict(), "{{ table.primary_key }}": item_id}
77+
78+ @app.delete("/{{ table.name | lower }}/{item_id}")
79+ async def delete_{{ table.name | lower }}(item_id: int):
80+ query = "DELETE FROM {{ table.name }} WHERE {{ table.primary_key }} = ?"
81+ execute_query(query, (item_id,))
82+ return {"message": "Item deleted successfully"}
83+ {% endfor %}"""
84+ self .template = Template (self .template_string )
85+
86+ def get_db_schema (self , db_path ):
87+ conn = sqlite3 .connect (db_path )
88+ cursor = conn .cursor ()
89+
90+ cursor .execute ("SELECT name FROM sqlite_master WHERE type='table';" )
91+ tables = cursor .fetchall ()
92+
93+ schema = []
94+ for table in tables :
95+ table_name = table [0 ]
96+ if table_name == 'sqlite_sequence' :
97+ continue # Skip this system table
98+
99+ cursor .execute (f"PRAGMA table_info({ table_name } )
100+ columns = cursor .fetchall ()
101+
102+ primary_key = next ((col [1 ] for col in columns if col [5 ] == 1 ), 'id' )
103+
104+ table_schema = {
105+ 'name' : table_name ,
106+ 'columns' : [],
107+ 'primary_key' : primary_key
108+ }
109+
110+ for col in columns :
111+ col_id , col_name , col_type , not_null , _ , _ = col
112+ py_type = self .map_sqlite_to_python_type (col_type , col_name )
113+ table_schema ['columns' ].append ({
114+ 'name' : col_name ,
115+ 'type' : col_type ,
116+ 'py_type' : py_type ,
117+ 'nullable' : not not_null
118+ })
119+
120+ schema .append (table_schema )
121+
122+ conn .close ()
123+ return schema
124+
125+ def map_sqlite_to_python_type (self , sqlite_type , col_name ):
126+ sqlite_type = sqlite_type .lower ()
127+ if 'int' in sqlite_type :
128+ return 'bool' if col_name .lower () in ['isadmin' , 'is_admin' ] else 'int'
129+ elif 'char' in sqlite_type or 'clob' in sqlite_type or 'text' in sqlite_type :
130+ return 'datetime' if 'date' in col_name .lower () or 'time' in col_name .lower () else 'str'
131+ elif 'real' in sqlite_type or 'floa' in sqlite_type or 'doub' in sqlite_type :
132+ return 'float'
133+ elif 'blob' in sqlite_type :
134+ return 'bytes'
135+ elif 'boolean' in sqlite_type :
136+ return 'bool'
137+ else :
138+ print (f"Unknown SQLite type: { sqlite_type } )
139+ return 'Any'
140+
141+ def generate_code (self , db_path ):
142+ schema = self .get_db_schema (db_path )
143+
144+ generated_files = {
145+ 'main.py' : self .env .get_template ('main.py.jinja' ).render (db_name = db_path ),
146+ 'database.py' : self .env .get_template ('database.py.jinja' ).render (db_name = db_path ),
147+ 'models.py' : self .env .get_template ('models.py.jinja' ).render (tables = schema ),
148+ 'routers/__init__.py' : '' ,
149+ }
150+
151+ for table in schema :
152+ router_file = f"routers/{ table ['name' ].lower ()}
153+ generated_files [router_file ] = self .env .get_template ('router.py.jinja' ).render (table = table )
154+
155+ return generated_files
0 commit comments