This repository was archived by the owner on Jun 5, 2025. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 90
Expand file tree
/
Copy pathinit_db.py
More file actions
43 lines (34 loc) · 1.38 KB
/
init_db.py
File metadata and controls
43 lines (34 loc) · 1.38 KB
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
import asyncio
from pathlib import Path
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
async def init_db():
"""Initialize the database with the schema."""
# Get the absolute path to the schema file
current_dir = Path(__file__).parent
schema_path = current_dir.parent.parent.parent / 'sql' / 'schema' / 'schema.sql'
if not schema_path.exists():
raise FileNotFoundError(f"Schema file not found at {schema_path}")
with open(schema_path, 'r') as f:
schema = f.read()
db_path = Path('codegate.db').absolute()
engine = create_async_engine(
f"sqlite+aiosqlite:///{db_path}",
echo=True, # We should toggle this to true when we do release codegate
isolation_level="AUTOCOMMIT", # Required for SQLite
)
try:
# Execute the schema
async with engine.begin() as conn:
# Split the schema into individual statements and execute each one
statements = [stmt.strip() for stmt in schema.split(';') if stmt.strip()]
for statement in statements:
# Use SQLAlchemy text() to create executable SQL statements
await conn.execute(text(statement))
finally:
await engine.dispose()
def init_db_sync():
"""Synchronous wrapper for init_db."""
asyncio.run(init_db())
if __name__ == '__main__':
init_db_sync()