from typing import Dict, List, Any import secrets class DatabaseTemplateManager: """Manage database templates and cloning""" def __init__(self): self.templates = self._load_templates() def _load_templates(self) -> Dict[str, Dict]: """Load pre-built database templates""" return { 'ecommerce': { 'name': 'E-commerce Store', 'description': 'Complete e-commerce database with products, orders, customers', 'icon': '🛒', 'tables': [ { 'name': 'products', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'description', 'type': 'TEXT'}, {'name': 'price', 'type': 'DECIMAL'}, {'name': 'stock', 'type': 'INTEGER'}, {'name': 'category', 'type': 'VARCHAR'}, {'name': 'image_url', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'customers', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'email', 'type': 'VARCHAR'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'phone', 'type': 'VARCHAR'}, {'name': 'address', 'type': 'TEXT'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'orders', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'customer_id', 'type': 'INTEGER'}, {'name': 'total', 'type': 'DECIMAL'}, {'name': 'status', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'order_items', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'order_id', 'type': 'INTEGER'}, {'name': 'product_id', 'type': 'INTEGER'}, {'name': 'quantity', 'type': 'INTEGER'}, {'name': 'price', 'type': 'DECIMAL'} ] } ] }, 'blog': { 'name': 'Blog Platform', 'description': 'Blog with posts, comments, categories, and tags', 'icon': '📝', 'tables': [ { 'name': 'posts', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'title', 'type': 'VARCHAR'}, {'name': 'slug', 'type': 'VARCHAR'}, {'name': 'content', 'type': 'TEXT'}, {'name': 'author_id', 'type': 'INTEGER'}, {'name': 'status', 'type': 'VARCHAR'}, {'name': 'published_at', 'type': 'TIMESTAMP'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'comments', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'post_id', 'type': 'INTEGER'}, {'name': 'author_name', 'type': 'VARCHAR'}, {'name': 'author_email', 'type': 'VARCHAR'}, {'name': 'content', 'type': 'TEXT'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'categories', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'slug', 'type': 'VARCHAR'} ] }, { 'name': 'tags', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'slug', 'type': 'VARCHAR'} ] } ] }, 'crm': { 'name': 'CRM System', 'description': 'Customer relationship management with leads, deals, contacts', 'icon': '👥', 'tables': [ { 'name': 'contacts', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'first_name', 'type': 'VARCHAR'}, {'name': 'last_name', 'type': 'VARCHAR'}, {'name': 'email', 'type': 'VARCHAR'}, {'name': 'phone', 'type': 'VARCHAR'}, {'name': 'company', 'type': 'VARCHAR'}, {'name': 'position', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'leads', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'contact_id', 'type': 'INTEGER'}, {'name': 'source', 'type': 'VARCHAR'}, {'name': 'status', 'type': 'VARCHAR'}, {'name': 'score', 'type': 'INTEGER'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'deals', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'contact_id', 'type': 'INTEGER'}, {'name': 'title', 'type': 'VARCHAR'}, {'name': 'value', 'type': 'DECIMAL'}, {'name': 'stage', 'type': 'VARCHAR'}, {'name': 'probability', 'type': 'INTEGER'}, {'name': 'close_date', 'type': 'DATE'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'activities', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'contact_id', 'type': 'INTEGER'}, {'name': 'type', 'type': 'VARCHAR'}, {'name': 'description', 'type': 'TEXT'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] } ] }, 'saas': { 'name': 'SaaS Application', 'description': 'Multi-tenant SaaS with users, subscriptions, billing', 'icon': '☁️', 'tables': [ { 'name': 'tenants', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'subdomain', 'type': 'VARCHAR'}, {'name': 'plan', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'users', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'tenant_id', 'type': 'INTEGER'}, {'name': 'email', 'type': 'VARCHAR'}, {'name': 'name', 'type': 'VARCHAR'}, {'name': 'role', 'type': 'VARCHAR'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'subscriptions', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'tenant_id', 'type': 'INTEGER'}, {'name': 'plan', 'type': 'VARCHAR'}, {'name': 'status', 'type': 'VARCHAR'}, {'name': 'started_at', 'type': 'TIMESTAMP'}, {'name': 'expires_at', 'type': 'TIMESTAMP'} ] }, { 'name': 'invoices', 'columns': [ {'name': 'id', 'type': 'INTEGER'}, {'name': 'tenant_id', 'type': 'INTEGER'}, {'name': 'amount', 'type': 'DECIMAL'}, {'name': 'status', 'type': 'VARCHAR'}, {'name': 'due_date', 'type': 'DATE'}, {'name': 'created_at', 'type': 'TIMESTAMP'} ] } ] } } def list_templates(self) -> List[Dict]: """List all available templates""" return [ { 'id': key, 'name': template['name'], 'description': template['description'], 'icon': template['icon'], 'tables_count': len(template['tables']) } for key, template in self.templates.items() ] def get_template(self, template_id: str) -> Dict: """Get template details""" template = self.templates.get(template_id) if not template: return {'ok': False, 'error': 'Template not found'} return {'ok': True, 'template': template} def create_from_template(self, template_id: str, database_name: str, user_store, username: str, workspace_id: str) -> Dict: """Create database from template""" from app.metadata import metadata from app.table_manager import table_manager template = self.templates.get(template_id) if not template: return {'ok': False, 'error': 'Template not found'} # Create database db_result = metadata.create_database(database_name, user_store) if not db_result.get('ok'): return db_result # Create tables created_tables = [] for table_def in template['tables']: result = table_manager.create( database_name, table_def['name'], table_def['columns'], user_store ) if result.get('ok'): created_tables.append(table_def['name']) return { 'ok': True, 'database': database_name, 'template': template_id, 'tables_created': created_tables } def clone_database(self, source_db: str, target_db: str, user_store, username: str, workspace_id: str) -> Dict: """Clone existing database""" from app.metadata import metadata from app.table_manager import table_manager from app.row_manager import row_manager # Create target database db_result = metadata.create_database(target_db, user_store) if not db_result.get('ok'): return db_result # Get all tables from source all_tables = table_manager.list(user_store) source_tables = [t for t in all_tables if t['database'] == source_db] cloned_tables = [] for table_info in source_tables: table_name = table_info['table'] columns = table_info['columns'] # Create table in target result = table_manager.create(target_db, table_name, columns, user_store) if not result.get('ok'): continue # Copy data rows = row_manager.read(source_db, table_name, user_store, limit=10000) if rows: import pandas as pd df = pd.DataFrame(rows) row_manager.write_df(target_db, table_name, df, username, user_store, workspace_id) cloned_tables.append(table_name) return { 'ok': True, 'source': source_db, 'target': target_db, 'tables_cloned': cloned_tables } database_template_manager = DatabaseTemplateManager()