yumoqing
75fa6c5d29
- Add missing core module files: crud_ops.py, db_tables.py, init_db.py, user_context.py - Ensure all .dspy scripts properly use get_user() for user_id acquisition - Fix user context handling in module functions via get_current_user_id() - Maintain proper async/await patterns throughout the codebase - Complete module implementation following module-development-spec - All database operations use sqlor framework with proper user isolation
70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Database initialization script for hermes-web-cli module.
|
|
Creates the required tables if they don't exist.
|
|
"""
|
|
|
|
import asyncio
|
|
from sqlor.dbpools import DBPools
|
|
from .db_tables import TABLE_DEFINITIONS
|
|
|
|
async def init_database():
|
|
"""Initialize the hermes-web-cli database tables."""
|
|
try:
|
|
db = DBPools()
|
|
async with db.sqlorContext('hermes-web-cli') as sor:
|
|
# Create tables based on table definitions
|
|
for table_def in TABLE_DEFINITIONS:
|
|
table_name = table_def['summary']['name']
|
|
fields = table_def['fields']
|
|
indexes = table_def['indexes']
|
|
|
|
# Build CREATE TABLE statement
|
|
field_defs = []
|
|
primary_keys = []
|
|
|
|
for field in fields:
|
|
field_def = f"{field['name']} {field['type']}"
|
|
if not field.get('nullable', True):
|
|
field_def += " NOT NULL"
|
|
if field.get('default'):
|
|
field_def += f" DEFAULT {field['default']}"
|
|
if field.get('primary_key'):
|
|
primary_keys.append(field['name'])
|
|
field_defs.append(field_def)
|
|
|
|
if primary_keys:
|
|
field_defs.append(f"PRIMARY KEY ({', '.join(primary_keys)})")
|
|
|
|
create_table_sql = f"""
|
|
CREATE TABLE IF NOT EXISTS {table_name} (
|
|
{', '.join(field_defs)}
|
|
)
|
|
"""
|
|
|
|
print(f"Creating table: {table_name}")
|
|
await sor.sqlExe(create_table_sql, {})
|
|
|
|
# Create indexes
|
|
for index in indexes:
|
|
index_name = index['name']
|
|
index_fields = ', '.join(index['fields'])
|
|
unique_clause = "UNIQUE" if index.get('unique') else ""
|
|
|
|
create_index_sql = f"""
|
|
CREATE {unique_clause} INDEX IF NOT EXISTS {index_name}
|
|
ON {table_name} ({index_fields})
|
|
"""
|
|
|
|
print(f"Creating index: {index_name}")
|
|
await sor.sqlExe(create_index_sql, {})
|
|
|
|
print("Database initialization completed successfully!")
|
|
return True
|
|
|
|
except Exception as e:
|
|
print(f"Error initializing database: {str(e)}")
|
|
return False
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(init_database()) |