|
| 1 | +"""Generator for Drizzle ORM TypeScript schema from SchemaForge IR.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +from ..ir import Schema, Table, Column, ColumnType, EnumType, Index |
| 5 | + |
| 6 | + |
| 7 | +# ColumnType -> Drizzle type function mapping |
| 8 | +_TYPE_TO_DRIZZLE: dict[ColumnType, str] = { |
| 9 | + ColumnType.STRING: "varchar", |
| 10 | + ColumnType.INTEGER: "integer", |
| 11 | + ColumnType.FLOAT: "real", |
| 12 | + ColumnType.BOOLEAN: "boolean", |
| 13 | + ColumnType.DATETIME: "timestamp", |
| 14 | + ColumnType.DATE: "date", |
| 15 | + ColumnType.TIME: "time", |
| 16 | + ColumnType.TEXT: "text", |
| 17 | + ColumnType.BLOB: "blob", |
| 18 | + ColumnType.JSON: "json", |
| 19 | + ColumnType.UUID: "uuid", |
| 20 | + ColumnType.ENUM: "varchar", # Handled via pgEnum separately |
| 21 | + ColumnType.DECIMAL: "numeric", |
| 22 | + ColumnType.CUSTOM: "text", # Fallback |
| 23 | +} |
| 24 | + |
| 25 | + |
| 26 | +class DrizzleGenerator: |
| 27 | + """Generate Drizzle ORM TypeScript schema from Schema IR.""" |
| 28 | + |
| 29 | + def __init__(self, dialect: str = "pg") -> None: |
| 30 | + """Initialize with dialect (pg, mysql, or sqlite).""" |
| 31 | + self.dialect = dialect |
| 32 | + |
| 33 | + def generate(self, schema: Schema) -> str: |
| 34 | + lines: list[str] = [] |
| 35 | + |
| 36 | + # Collect imports |
| 37 | + imports = self._collect_imports(schema) |
| 38 | + lines.append(imports) |
| 39 | + lines.append("") |
| 40 | + |
| 41 | + # Generate enums (PostgreSQL only) |
| 42 | + if self.dialect == "pg" and schema.enums: |
| 43 | + for enum in schema.enums: |
| 44 | + lines.append(self._generate_enum(enum)) |
| 45 | + lines.append("") |
| 46 | + |
| 47 | + # Generate tables |
| 48 | + for table in schema.tables: |
| 49 | + lines.append(self._generate_table(table, schema.enums)) |
| 50 | + lines.append("") |
| 51 | + |
| 52 | + return "\n".join(lines) |
| 53 | + |
| 54 | + def _collect_imports(self, schema: Schema) -> str: |
| 55 | + """Generate the import statements.""" |
| 56 | + table_func = f"{self.dialect}Table" |
| 57 | + type_imports: set[str] = {table_func} |
| 58 | + dialect_prefix = self.dialect |
| 59 | + |
| 60 | + # Determine dialect-specific import path |
| 61 | + if self.dialect == "pg": |
| 62 | + import_path = "drizzle-orm/pg-core" |
| 63 | + elif self.dialect == "mysql": |
| 64 | + import_path = "drizzle-orm/mysql-core" |
| 65 | + else: |
| 66 | + import_path = "drizzle-orm/sqlite-core" |
| 67 | + |
| 68 | + # Collect type function names needed |
| 69 | + for table in schema.tables: |
| 70 | + for col in table.columns: |
| 71 | + drizzle_type = self._get_drizzle_type(col) |
| 72 | + type_imports.add(drizzle_type) |
| 73 | + |
| 74 | + # Add serial for primary keys with integer type |
| 75 | + if col.primary_key and col.type == ColumnType.INTEGER and not col.default: |
| 76 | + type_imports.add("serial") |
| 77 | + type_imports.discard("integer") |
| 78 | + |
| 79 | + # Add pgEnum if needed |
| 80 | + if self.dialect == "pg" and schema.enums: |
| 81 | + type_imports.add("pgEnum") |
| 82 | + |
| 83 | + imports_list = sorted(type_imports) |
| 84 | + return f"import {{ {', '.join(imports_list)} }} from '{import_path}';" |
| 85 | + |
| 86 | + def _get_drizzle_type(self, col: Column) -> str: |
| 87 | + """Get the Drizzle type function name for a column.""" |
| 88 | + if col.primary_key and col.type == ColumnType.INTEGER and not col.default: |
| 89 | + return "serial" |
| 90 | + |
| 91 | + # Use varchar for string types with length |
| 92 | + if col.type == ColumnType.STRING: |
| 93 | + return "varchar" |
| 94 | + |
| 95 | + return _TYPE_TO_DRIZZLE.get(col.type, "text") |
| 96 | + |
| 97 | + def _generate_enum(self, enum: EnumType) -> str: |
| 98 | + """Generate a pgEnum declaration.""" |
| 99 | + values = ", ".join(f"'{v}'" for v in enum.values) |
| 100 | + return f"export const {enum.name} = pgEnum('{enum.name}', [{values}]);" |
| 101 | + |
| 102 | + def _generate_table(self, table: Table, enums: list[EnumType]) -> str: |
| 103 | + """Generate a table definition.""" |
| 104 | + table_func = f"{self.dialect}Table" |
| 105 | + # Use table name as variable name (camelCase) |
| 106 | + var_name = self._to_camel_case(table.name) |
| 107 | + |
| 108 | + lines = [f"export const {var_name} = {table_func}('{table.name}', {{"] |
| 109 | + |
| 110 | + enum_names = {e.name for e in enums} |
| 111 | + |
| 112 | + for col in table.columns: |
| 113 | + col_line = self._generate_column(col, enum_names) |
| 114 | + lines.append(f" {col_line}") |
| 115 | + |
| 116 | + lines.append("});") |
| 117 | + return "\n".join(lines) |
| 118 | + |
| 119 | + def _generate_column(self, col: Column, enum_names: set[str]) -> str: |
| 120 | + """Generate a single column definition.""" |
| 121 | + # Determine DB column name (from comment if stored) |
| 122 | + db_name = col.name |
| 123 | + if col.comment and col.comment.startswith("db_name:"): |
| 124 | + db_name = col.comment.split(":", 1)[1] |
| 125 | + |
| 126 | + # Type function call |
| 127 | + type_func = self._get_drizzle_type(col) |
| 128 | + |
| 129 | + # Build type arguments |
| 130 | + type_args = self._build_type_args(col, type_func) |
| 131 | + |
| 132 | + # Use enum type if this is an enum column |
| 133 | + if col.type == ColumnType.ENUM and col.custom_type in enum_names: |
| 134 | + # Reference the pgEnum variable |
| 135 | + base = f"{col.custom_type}('{db_name}')" |
| 136 | + elif type_args: |
| 137 | + base = f"{type_func}('{db_name}', {{ {type_args} }})" |
| 138 | + else: |
| 139 | + base = f"{type_func}('{db_name}')" |
| 140 | + |
| 141 | + # Build chain methods |
| 142 | + chain = [] |
| 143 | + if not col.primary_key: |
| 144 | + if not col.nullable: |
| 145 | + chain.append(".notNull()") |
| 146 | + if col.unique: |
| 147 | + chain.append(".unique()") |
| 148 | + |
| 149 | + if col.default is not None: |
| 150 | + if col.default == "now()": |
| 151 | + chain.append(".defaultNow()") |
| 152 | + elif isinstance(col.default, bool): |
| 153 | + chain.append(f".default({str(col.default).lower()})") |
| 154 | + elif isinstance(col.default, str): |
| 155 | + chain.append(f".default('{col.default}')") |
| 156 | + elif isinstance(col.default, (int, float)): |
| 157 | + chain.append(f".default({col.default})") |
| 158 | + else: |
| 159 | + chain.append(f".default({col.default})") |
| 160 | + |
| 161 | + if col.primary_key: |
| 162 | + chain.append(".primaryKey()") |
| 163 | + |
| 164 | + # Format: fieldName: typeFunc('name').notNull().primaryKey() |
| 165 | + result = f"{col.name}: {base}{''.join(chain)}" |
| 166 | + return result + "," |
| 167 | + |
| 168 | + def _build_type_args(self, col: Column, type_func: str) -> str: |
| 169 | + """Build type-specific arguments like { length: 255 }.""" |
| 170 | + args = [] |
| 171 | + if "length" in col.type_args: |
| 172 | + args.append(f"length: {col.type_args['length']}") |
| 173 | + if "precision" in col.type_args: |
| 174 | + args.append(f"precision: {col.type_args['precision']}") |
| 175 | + if "scale" in col.type_args: |
| 176 | + args.append(f"scale: {col.type_args['scale']}") |
| 177 | + |
| 178 | + return ", ".join(args) # Join with ", " for proper spacing |
| 179 | + |
| 180 | + def _to_camel_case(self, name: str) -> str: |
| 181 | + """Convert snake_case table name to camelCase variable name.""" |
| 182 | + parts = name.split("_") |
| 183 | + return parts[0] + "".join(p.capitalize() for p in parts[1:]) |
0 commit comments