diff --git a/fixtures/sample.alembic.py b/fixtures/sample.alembic.py index a4cf28b..8cebc2a 100644 --- a/fixtures/sample.alembic.py +++ b/fixtures/sample.alembic.py @@ -4,24 +4,26 @@ Revises: Create Date: 2026-05-15 03:00:00.000000 """ + import sqlalchemy as sa from alembic import op # revision identifiers, used by Alembic. -revision = 'sample' +revision = "sample" down_revision = None def upgrade() -> None: - op.create_table('users', - sa.Column('id', sa.Integer(), primary_key=True), - sa.Column('name', sa.String(100), nullable=False), - sa.Column('email', sa.String(255), nullable=False, unique=True), - sa.Column('role', sa.Enum('admin', 'editor', 'viewer'), nullable=False), - sa.Column('is_active', sa.Boolean(), server_default=True), - sa.Column('created_at', sa.DateTime(), server_default=sa.func.now()), + op.create_table( + "users", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("name", sa.String(100), nullable=False), + sa.Column("email", sa.String(255), nullable=False, unique=True), + sa.Column("role", sa.Enum("admin", "editor", "viewer"), nullable=False), + sa.Column("is_active", sa.Boolean(), server_default=True), + sa.Column("created_at", sa.DateTime(), server_default=sa.func.now()), ) def downgrade() -> None: - op.drop_table('users') + op.drop_table("users") diff --git a/fixtures/sample.django.py b/fixtures/sample.django.py index dac47f9..f215c9c 100644 --- a/fixtures/sample.django.py +++ b/fixtures/sample.django.py @@ -7,10 +7,10 @@ class UserRole(models.TextChoices): - ADMIN = 'admin', 'Admin' - EDITOR = 'editor', 'Editor' - AUTHOR = 'author', 'Author' - SUBSCRIBER = 'subscriber', 'Subscriber' + ADMIN = "admin", "Admin" + EDITOR = "editor", "Editor" + AUTHOR = "author", "Author" + SUBSCRIBER = "subscriber", "Subscriber" class User(models.Model): @@ -26,7 +26,7 @@ class User(models.Model): created_at = models.DateTimeField(auto_now_add=True) class Meta: - db_table = 'users' + db_table = "users" def __str__(self): return self.name @@ -38,8 +38,8 @@ class Category(models.Model): sort_order = models.IntegerField(default=0) class Meta: - db_table = 'categories' - verbose_name_plural = 'categories' + db_table = "categories" + verbose_name_plural = "categories" def __str__(self): return self.name @@ -50,19 +50,19 @@ class Post(models.Model): slug = models.SlugField(max_length=255, unique=True) content = models.TextField() excerpt = models.TextField(null=True, blank=True, max_length=500) - status = models.CharField(max_length=20, default='draft') + status = models.CharField(max_length=20, default="draft") published_at = models.DateTimeField(null=True, blank=True) author = models.ForeignKey( User, on_delete=models.CASCADE, - related_name='posts', + related_name="posts", ) category = models.ForeignKey( Category, on_delete=models.SET_NULL, null=True, blank=True, - related_name='posts', + related_name="posts", ) view_count = models.IntegerField(default=0) rating = models.DecimalField(max_digits=3, decimal_places=2, default=0.00) @@ -70,7 +70,7 @@ class Post(models.Model): updated_at = models.DateTimeField(auto_now=True) class Meta: - db_table = 'posts' + db_table = "posts" def __str__(self): return self.title diff --git a/pyproject.toml b/pyproject.toml index 140f4f5..46f2978 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -37,10 +37,10 @@ dependencies = [ dev = [ "pytest>=7.0", "pytest-cov>=4.0", - "mcp>=1.0", + "mcp>=1.0,<2", "ruff>=0.4.0", ] -mcp = ["mcp>=1.0"] +mcp = ["mcp>=1.0,<2"] [project.scripts] schemaforge = "schemaforge.cli:main" diff --git a/scripts/check_consistency.py b/scripts/check_consistency.py index d769b2a..74801b7 100644 --- a/scripts/check_consistency.py +++ b/scripts/check_consistency.py @@ -1,4 +1,5 @@ """Schema consistency check script for CI/CD pipelines.""" + import os import sys diff --git a/src/schemaforge/check.py b/src/schemaforge/check.py index 0f82245..32eefd6 100644 --- a/src/schemaforge/check.py +++ b/src/schemaforge/check.py @@ -73,9 +73,7 @@ def check_directory( failures.append(f" FAIL {name}: {e}") if not converted: - return "FAIL: No files could be converted to {}\n{}".format( - canonical, "\n".join(failures) - ) + return "FAIL: No files could be converted to {}\n{}".format(canonical, "\n".join(failures)) # Compare pairwise mismatches: list[str] = [] diff --git a/src/schemaforge/diff.py b/src/schemaforge/diff.py index 4fcdcf1..30ac654 100644 --- a/src/schemaforge/diff.py +++ b/src/schemaforge/diff.py @@ -75,13 +75,7 @@ def diff_schemas(text_a: str, text_b: str, fmt: str) -> str: ) for name in common ) - if ( - not added - and not removed - and not has_table_diffs - and not enum_added - and not enum_removed - ): + if not added and not removed and not has_table_diffs and not enum_added and not enum_removed: lines.append("No differences found.") return "\n".join(lines) diff --git a/src/schemaforge/generators/alembic_generator.py b/src/schemaforge/generators/alembic_generator.py index a345899..3ca705e 100644 --- a/src/schemaforge/generators/alembic_generator.py +++ b/src/schemaforge/generators/alembic_generator.py @@ -73,9 +73,7 @@ def generate( lines.append("# ### enum type definitions ###") for enum_type in schema.enums: values = ", ".join(f"'{v}'" for v in enum_type.values) - lines.append( - f'op.execute("CREATE TYPE {enum_type.name} AS ENUM ({values})")' - ) + lines.append(f'op.execute("CREATE TYPE {enum_type.name} AS ENUM ({values})")') lines.append("") lines.append("# revision identifiers, used by Alembic.") @@ -107,9 +105,7 @@ def generate( for table in reversed(schema.tables): for idx in reversed(table.indexes): idx_name = idx.name or f"idx_{'_'.join(idx.columns)}" - lines.append( - f" op.drop_index('{idx_name}', table_name='{table.name}')" - ) + lines.append(f" op.drop_index('{idx_name}', table_name='{table.name}')") for table in reversed(schema.tables): lines.append(f" op.drop_table('{table.name}')") for enum_type in reversed(schema.enums): @@ -139,14 +135,9 @@ def _generate_table(self, table: Table) -> tuple[list[str], list[str]]: col_list = ", ".join(f"'{c}'" for c in idx.columns) idx_name = idx.name or f"idx_{'_'.join(idx.columns)}" if idx.unique: - index_lines.append( - f" op.create_unique_constraint('{idx_name}', " - f"'{table.name}', [{col_list}])" - ) + index_lines.append(f" op.create_unique_constraint('{idx_name}', '{table.name}', [{col_list}])") else: - index_lines.append( - f" op.create_index('{idx_name}', '{table.name}', [{col_list}])" - ) + index_lines.append(f" op.create_index('{idx_name}', '{table.name}', [{col_list}])") return table_lines, index_lines @@ -176,16 +167,12 @@ def _column_def(self, col: Column) -> str: kwargs.append("unique=True") # fn: defaults (server_default) - fn_default = resolve_fn_default( - col, fn_wrapper="sa.func.{}", expr_fallback="sa.text('{}')" - ) + fn_default = resolve_fn_default(col, fn_wrapper="sa.func.{}", expr_fallback="sa.text('{}')") if fn_default: kwargs.append(f"server_default={fn_default}") # Literal defaults - if col.default is not None and not ( - isinstance(col.default, str) and col.default.startswith("fn:") - ): + if col.default is not None and not (isinstance(col.default, str) and col.default.startswith("fn:")): lit = format_literal_default(col) kwargs.append(f"server_default={lit}") diff --git a/src/schemaforge/generators/django_generator.py b/src/schemaforge/generators/django_generator.py index 8a49a7e..dbbb298 100644 --- a/src/schemaforge/generators/django_generator.py +++ b/src/schemaforge/generators/django_generator.py @@ -69,10 +69,7 @@ def _generate_model(self, table) -> str: # Unique on single column is already handled via field option pass else: - lines.append( - f" {col} = models.{'IntegerField' if col else 'CharField'}(" - f"db_index=True)" - ) + lines.append(f" {col} = models.{'IntegerField' if col else 'CharField'}(db_index=True)") # Unique constraints (multi-column) unique_sets: list[list[str]] = [] @@ -103,9 +100,7 @@ def _generate_model(self, table) -> str: def _field_def(self, col: Column) -> str: """Generate a Django model field definition.""" - django_field = resolve_type( - col, self._FIELD_MAP, fmt="django", type_config=self._type_config - ) + django_field = resolve_type(col, self._FIELD_MAP, fmt="django", type_config=self._type_config) if not django_field.endswith("Field"): django_field = django_field + "Field" diff --git a/src/schemaforge/generators/drizzle_generator.py b/src/schemaforge/generators/drizzle_generator.py index eabc2e7..79b5eda 100644 --- a/src/schemaforge/generators/drizzle_generator.py +++ b/src/schemaforge/generators/drizzle_generator.py @@ -27,9 +27,7 @@ class DrizzleGenerator: """Generate Drizzle ORM TypeScript schema from Schema IR.""" - def __init__( - self, dialect: str = "pg", type_config: TypeConfig | None = None - ) -> None: + def __init__(self, dialect: str = "pg", type_config: TypeConfig | None = None) -> None: """Initialize with dialect and optional custom type overrides. Args: @@ -81,11 +79,7 @@ def _collect_imports(self, schema: Schema) -> str: type_imports.add(drizzle_type) # Add serial for primary keys with integer type - if ( - col.primary_key - and col.type == ColumnType.INTEGER - and not col.default - ): + if col.primary_key and col.type == ColumnType.INTEGER and not col.default: type_imports.add("serial") type_imports.discard("integer") diff --git a/src/schemaforge/generators/json_schema_generator.py b/src/schemaforge/generators/json_schema_generator.py index 27e140a..0b9c29b 100644 --- a/src/schemaforge/generators/json_schema_generator.py +++ b/src/schemaforge/generators/json_schema_generator.py @@ -163,9 +163,7 @@ def _column_to_prop(self, col: Column) -> dict[str, Any]: return self._add_base_annotations(prop, col) - def _add_base_annotations( - self, prop: dict[str, Any], col: Column - ) -> dict[str, Any]: + def _add_base_annotations(self, prop: dict[str, Any], col: Column) -> dict[str, Any]: """Add common annotations (description, default, comment) to a property.""" if col.comment: prop["description"] = col.comment diff --git a/src/schemaforge/generators/prisma_generator.py b/src/schemaforge/generators/prisma_generator.py index 6bea711..279cd76 100644 --- a/src/schemaforge/generators/prisma_generator.py +++ b/src/schemaforge/generators/prisma_generator.py @@ -34,10 +34,7 @@ def generate(self, schema: Schema) -> str: """Generate Prisma schema from IR.""" parts: list[str] = [] parts.append('generator client {\n provider = "prisma-client-js"\n}') - parts.append( - 'datasource db {\n provider = "postgresql"\n' - ' url = env("DATABASE_URL")\n}' - ) + parts.append('datasource db {\n provider = "postgresql"\n url = env("DATABASE_URL")\n}') # Generate enums for enum_type in schema.enums: @@ -65,9 +62,7 @@ def _generate_model(self, table) -> str: def _field_def(self, col: Column) -> str: """Generate a Prisma field definition.""" - prisma_type = resolve_type( - col, self._TYPE_MAP, fmt="prisma", type_config=self._type_config - ) + prisma_type = resolve_type(col, self._TYPE_MAP, fmt="prisma", type_config=self._type_config) # Handle String with length (Prisma uses @db.VarChar) — skip if overridden if ( diff --git a/src/schemaforge/generators/sqlalchemy_generator.py b/src/schemaforge/generators/sqlalchemy_generator.py index 49deea7..df29dd0 100644 --- a/src/schemaforge/generators/sqlalchemy_generator.py +++ b/src/schemaforge/generators/sqlalchemy_generator.py @@ -116,10 +116,7 @@ def _column_def(self, col: Column) -> tuple[list[str], set[str]]: # Primary key if col.primary_key: kwargs.append("primary_key=True") - if ( - col.type != ColumnType.INTEGER - or col.type_args.get("autoincrement") is False - ): + if col.type != ColumnType.INTEGER or col.type_args.get("autoincrement") is False: kwargs.append("autoincrement=False") # Nullable — SQLAlchemy defaults to True, so only emit when False @@ -135,16 +132,12 @@ def _column_def(self, col: Column) -> tuple[list[str], set[str]]: kwargs.append("index=True") # fn: defaults (server_default) - fn_default = resolve_fn_default( - col, fn_wrapper="func.{}", expr_fallback="text('{}')" - ) + fn_default = resolve_fn_default(col, fn_wrapper="func.{}", expr_fallback="text('{}')") if fn_default: kwargs.append(f"server_default={fn_default}") # Literal defaults - if col.default is not None and not ( - isinstance(col.default, str) and col.default.startswith("fn:") - ): + if col.default is not None and not (isinstance(col.default, str) and col.default.startswith("fn:")): lit = format_literal_default(col) if isinstance(col.default, bool | int | float | str): kwargs.append(f"default={lit}") diff --git a/src/schemaforge/generators/typeorm_generator.py b/src/schemaforge/generators/typeorm_generator.py index 5fbcde8..e287899 100644 --- a/src/schemaforge/generators/typeorm_generator.py +++ b/src/schemaforge/generators/typeorm_generator.py @@ -61,21 +61,14 @@ def generate(self, schema: Schema) -> str: has_index = any(t.indexes for t in schema.tables) if has_index: imports.append("Index") - has_unique = any( - any(c.unique and not c.primary_key for c in t.columns) - for t in schema.tables - ) + has_unique = any(any(c.unique and not c.primary_key for c in t.columns) for t in schema.tables) if has_unique: imports.append("Unique") if len(imports) == 1 and imports[0] == "Entity": - parts[0] = ( - 'import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";' - ) + parts[0] = 'import { Entity, PrimaryGeneratedColumn, Column } from "typeorm";' else: - parts[0] = ( - "import { " + ", ".join(sorted(set(imports))) + ' } from "typeorm";' - ) + parts[0] = "import { " + ", ".join(sorted(set(imports))) + ' } from "typeorm";' entities: list[str] = [] for table in schema.tables: @@ -121,9 +114,7 @@ def _column_def(self, col: Column) -> list[str]: options: dict[str, str] = {} # Determine TypeORM type - col_type = resolve_type( - col, self._TYPE_MAP, fmt="typeorm", type_config=self._type_config - ) + col_type = resolve_type(col, self._TYPE_MAP, fmt="typeorm", type_config=self._type_config) # Primary key handling if col.primary_key: diff --git a/src/schemaforge/mcp_server.py b/src/schemaforge/mcp_server.py index c7a7f0c..308c833 100644 --- a/src/schemaforge/mcp_server.py +++ b/src/schemaforge/mcp_server.py @@ -24,26 +24,29 @@ FastMCP = None # type: ignore -def _confined_directory(directory: str) -> Path: - """Resolve *directory* and confirm it stays within the allowed root. +def _confined_path(path: str, *, kind: str = "Path") -> Path: + """Resolve *path* and confirm it stays within the allowed root. - The ``check`` tool iterates and reads files under the given directory. To - keep an AI agent (or, in SSE mode, a remote caller) from reading arbitrary - locations on the host, requests are confined to a root — the + MCP directory checks and type-map reads are confined to a root - the ``SCHEMAFORGE_MCP_ROOT`` environment variable if set, otherwise the current - working directory the server was launched in. Escaping the root raises - ``PermissionError``. + working directory. Canonical resolution follows links before validation. + Escaping the root raises ``PermissionError``. """ root = Path(os.environ.get("SCHEMAFORGE_MCP_ROOT", Path.cwd())).resolve() - target = Path(directory).resolve() + target = Path(path).resolve() if target != root and not target.is_relative_to(root): raise PermissionError( - f"Directory '{directory}' is outside the allowed root '{root}'. " + f"{kind} '{path}' is outside the allowed root '{root}'. " f"Set SCHEMAFORGE_MCP_ROOT to permit a different base directory." ) return target +def _confined_directory(directory: str) -> Path: + """Resolve a check directory within the allowed MCP root.""" + return _confined_path(directory, kind="Directory") + + # All supported formats _FORMATS = [ "sql", @@ -73,15 +76,12 @@ def _confined_directory(directory: str) -> Path: } -def create_server() -> Any: +def create_server(*, host: str = "127.0.0.1", port: int = 8000) -> Any: """Create and configure the MCP server with all SchemaForge tools.""" if FastMCP is None: - raise ImportError( - "The 'mcp' package is required to run the MCP server.\n" - "Install it with: pip install mcp" - ) + raise ImportError("The 'mcp' package is required to run the MCP server.\nInstall it with: pip install mcp") - server = FastMCP("SchemaForge", log_level="WARNING") + server = FastMCP("SchemaForge", log_level="WARNING", host=host, port=port) @server.tool( name="convert", @@ -102,7 +102,7 @@ def convert_tool( from_format: Source format (sql, prisma, drizzle, typeorm, django, sqlalchemy, alembic, json_schema, graphql, ef, scala). to_format: Target format (same options as from_format). - type_map_path: Optional path to a YAML/JSON type mapping config file. + type_map_path: Optional YAML/JSON type map within the allowed MCP root. """ if from_format not in _FORMATS: return f"Error: Unsupported source format '{from_format}'. Supported: {', '.join(_FORMATS)}" @@ -112,14 +112,12 @@ def convert_tool( type_config: TypeConfig | None = None if type_map_path: try: - type_config = TypeConfig.from_file(type_map_path) - except (FileNotFoundError, ValueError) as e: + type_config = TypeConfig.from_file(_confined_path(type_map_path, kind="Type map")) + except (FileNotFoundError, PermissionError, ValueError) as e: return f"Error loading type map: {e}" try: - result = convert_schema( - schema_text, from_format, to_format, type_config=type_config - ) + result = convert_schema(schema_text, from_format, to_format, type_config=type_config) return result except ValueError as e: return f"Error: {e}" @@ -151,11 +149,7 @@ def diff_tool( try: result = diff_schemas(schema_a, schema_b, format) - return ( - result - if result.strip() - else "No differences found — schemas are equivalent." - ) + return result if result.strip() else "No differences found — schemas are equivalent." except Exception as e: return f"Error: {e}" @@ -174,13 +168,12 @@ def check_tool( Args: directory: Path to directory containing schema files. canonical: Canonical format for comparison (default: sql). - type_map_path: Optional path to a YAML/JSON type mapping config file. + type_map_path: Optional YAML/JSON type map within the allowed MCP root. """ try: safe_dir = _confined_directory(directory) - result = check_directory( - str(safe_dir), canonical=canonical, type_map_path=type_map_path - ) + safe_type_map = str(_confined_path(type_map_path, kind="Type map")) if type_map_path else None + result = check_directory(str(safe_dir), canonical=canonical, type_map_path=safe_type_map) return result except PermissionError as e: return f"Error: {e}" @@ -247,10 +240,10 @@ def mcp_command(sse: bool, host: str, port: int) -> None: By default runs in stdio mode for AI clients (Claude Desktop, Cursor, etc.). Use --sse for HTTP transport. """ - server = create_server() + server = create_server(host=host, port=port) if sse else create_server() if sse: click.echo(f"Starting SchemaForge MCP server on http://{host}:{port}", err=True) - server.run(transport="sse", host=host, port=port) + server.run(transport="sse") else: server.run(transport="stdio") diff --git a/src/schemaforge/parsers/django_parser.py b/src/schemaforge/parsers/django_parser.py index 99b8f92..a3cca73 100644 --- a/src/schemaforge/parsers/django_parser.py +++ b/src/schemaforge/parsers/django_parser.py @@ -122,9 +122,7 @@ def _parse_model(self, name: str, body: str) -> Table | None: field_class = field_m.group(2) # Extract args between the outermost parens - args_start = stripped.index( - "(", stripped.index(field_class) + len(field_class) - ) + args_start = stripped.index("(", stripped.index(field_class) + len(field_class)) paren_depth = 0 args_end = len(stripped) for j, ch in enumerate(stripped[args_start:]): diff --git a/src/schemaforge/parsers/drizzle_parser.py b/src/schemaforge/parsers/drizzle_parser.py index f8f46fd..0019139 100644 --- a/src/schemaforge/parsers/drizzle_parser.py +++ b/src/schemaforge/parsers/drizzle_parser.py @@ -85,9 +85,7 @@ def parse(self, text: str) -> Schema: m.group(1) enum_name = m.group(2) values_str = m.group(3) - values = [ - v.strip().strip("'\"") for v in values_str.split(",") if v.strip() - ] + values = [v.strip().strip("'\"") for v in values_str.split(",") if v.strip()] schema.enums.append(EnumType(name=enum_name, values=values)) for m in table_pattern.finditer(text): diff --git a/src/schemaforge/parsers/ef_parser.py b/src/schemaforge/parsers/ef_parser.py index 4088f22..342b4ef 100644 --- a/src/schemaforge/parsers/ef_parser.py +++ b/src/schemaforge/parsers/ef_parser.py @@ -184,9 +184,7 @@ def _property_to_column( if ann_args and "=" not in ann_args: pass # Column name override — skip for now elif ann_name == "Index": - is_unique = "Unique" in ann_args or "IsUnique=true" in ann_args.replace( - " ", "" - ) + is_unique = "Unique" in ann_args or "IsUnique=true" in ann_args.replace(" ", "") elif ann_name == "Table": pass # Already handled at class level diff --git a/src/schemaforge/parsers/prisma_parser.py b/src/schemaforge/parsers/prisma_parser.py index 5f34d2d..d9eba20 100644 --- a/src/schemaforge/parsers/prisma_parser.py +++ b/src/schemaforge/parsers/prisma_parser.py @@ -67,12 +67,7 @@ def _parse_model(self, name: str, lines: list[str]) -> Table | None: for line in lines: # Skip decorators, comments, and block-level attributes stripped = line.strip() - if ( - not stripped - or stripped.startswith("@@") - or stripped.startswith("//") - or stripped.startswith("#") - ): + if not stripped or stripped.startswith("@@") or stripped.startswith("//") or stripped.startswith("#"): continue # Parse field: name type [modifiers...] @@ -97,9 +92,7 @@ def _parse_model(self, name: str, lines: list[str]) -> Table | None: type_args = {} if col_type == ColumnType.STRING and "@db.VarChar" in constraints: # Only capture length if explicitly constrained - db_match = __import__("re").search( - r"@db\.VarChar\((\d+)\)", constraints - ) + db_match = __import__("re").search(r"@db\.VarChar\((\d+)\)", constraints) if db_match: type_args["length"] = int(db_match.group(1)) diff --git a/src/schemaforge/parsers/scala_parser.py b/src/schemaforge/parsers/scala_parser.py index 44f71da..aa12102 100644 --- a/src/schemaforge/parsers/scala_parser.py +++ b/src/schemaforge/parsers/scala_parser.py @@ -91,9 +91,7 @@ def _parse_default(value_str: str) -> Any: return None # Quoted strings - if (val.startswith('"') and val.endswith('"')) or ( - val.startswith('"""') and val.endswith('"""') - ): + if (val.startswith('"') and val.endswith('"')) or (val.startswith('"""') and val.endswith('"""')): inner = val.strip('"') # Handle interpolation: s"..." idx = inner.find("$") @@ -143,9 +141,7 @@ def parse(self, text: str) -> Schema: fields = _FIELD_RE.findall(params_str) for field_name, field_type, default_str in fields: - col = self._field_to_column( - field_name, field_type.strip(), default_str.strip() - ) + col = self._field_to_column(field_name, field_type.strip(), default_str.strip()) if col: table.columns.append(col) @@ -154,9 +150,7 @@ def parse(self, text: str) -> Schema: return schema - def _field_to_column( - self, name: str, raw_type: str, default_str: str - ) -> Column | None: + def _field_to_column(self, name: str, raw_type: str, default_str: str) -> Column | None: """Convert a Scala field to a Column IR.""" clean_type, is_optional = _clean_scala_type(raw_type) diff --git a/src/schemaforge/parsers/sql_parser.py b/src/schemaforge/parsers/sql_parser.py index 6e40541..6adcf78 100644 --- a/src/schemaforge/parsers/sql_parser.py +++ b/src/schemaforge/parsers/sql_parser.py @@ -245,11 +245,7 @@ def _parse_column_def(self, defn: str) -> Column | None: return None col_name = tokens[0] - if ( - col_name.startswith('"') - or col_name.startswith("`") - or col_name.startswith("[") - ): + if col_name.startswith('"') or col_name.startswith("`") or col_name.startswith("["): col_name = col_name.strip('"`[]') # Find the type (skip quoted name) @@ -264,20 +260,14 @@ def _parse_column_def(self, defn: str) -> Column | None: type_start += 1 type_raw += " " + tokens[type_start] - type_name = ( - type_raw.split("(")[0].upper() if "(" in type_raw else type_raw.upper() - ) + type_name = type_raw.split("(")[0].upper() if "(" in type_raw else type_raw.upper() col_type = self._TYPE_MAP.get(type_name, ColumnType.CUSTOM) # Extract type args type_args: dict[str, Any] = {} if "(" in type_raw: - args_str = ( - type_raw[type_raw.index("(") + 1 : type_raw.index(")")] - if ")" in type_raw - else "" - ) + args_str = type_raw[type_raw.index("(") + 1 : type_raw.index(")")] if ")" in type_raw else "" if type_name == "ENUM": # Inline ENUM('a','b','c') — extract values via regex to handle multi-token args type_args["values"] = re.findall(r"'([^']*)'", args_str) @@ -294,9 +284,7 @@ def _parse_column_def(self, defn: str) -> Column | None: type_args["scale"] = int(parts[1]) # Parse constraints - constraints = ( - " ".join(tokens[type_start + 1 :]) if type_start + 1 < len(tokens) else "" - ) + constraints = " ".join(tokens[type_start + 1 :]) if type_start + 1 < len(tokens) else "" is_pk = "PRIMARY KEY" in constraints.upper() is_not_null = "NOT NULL" in constraints.upper() @@ -329,9 +317,7 @@ def _parse_column_def(self, defn: str) -> Column | None: is_fn = ( upper_val in self._SQL_FN_KEYWORDS or upper_val.rstrip("()") in self._SQL_FN_KEYWORDS - or re.match( - r"^\w+\(", val - ) # Any function call: nextval(), now(), etc. + or re.match(r"^\w+\(", val) # Any function call: nextval(), now(), etc. ) if is_fn: col.default = f"fn:{val}" @@ -365,9 +351,7 @@ def _parse_index_definition(self, defn: str) -> Index | None: def _parse_create_enum(self, stmt: str) -> EnumType | None: """Parse a CREATE TYPE ... AS ENUM statement.""" - m = re.match( - r"CREATE\s+TYPE\s+(\w+)\s+AS\s+ENUM\s*\(([^)]+)\)", stmt, re.IGNORECASE - ) + m = re.match(r"CREATE\s+TYPE\s+(\w+)\s+AS\s+ENUM\s*\(([^)]+)\)", stmt, re.IGNORECASE) if m: name = m.group(1) values = re.findall(r"'([^']*)'", m.group(2)) @@ -410,9 +394,7 @@ def _parse_table_options(self, after_paren: str) -> dict[str, str]: key_m = re.match(r"(\w+)\s*=\s*", remaining) if not key_m: # Try COMMENT 'xxx' (no equals) - comment_m = re.match( - r"COMMENT\s+'([^']*)'", remaining, re.IGNORECASE - ) + comment_m = re.match(r"COMMENT\s+'([^']*)'", remaining, re.IGNORECASE) if comment_m: options["COMMENT"] = comment_m.group(1) i += comment_m.end() diff --git a/src/schemaforge/parsers/sqlalchemy_parser.py b/src/schemaforge/parsers/sqlalchemy_parser.py index bf20818..81b01f3 100644 --- a/src/schemaforge/parsers/sqlalchemy_parser.py +++ b/src/schemaforge/parsers/sqlalchemy_parser.py @@ -101,11 +101,7 @@ class Product(DeclarativeBase): continue # Check if any base looks like a declarative base bases = [b.strip() for b in base_name.split(",")] - is_model = any( - b in ("Base", "DeclarativeBase", "DeclarativeMeta") - or b.endswith("Base") - for b in bases - ) + is_model = any(b in ("Base", "DeclarativeBase", "DeclarativeMeta") or b.endswith("Base") for b in bases) if not is_model: i += 1 continue @@ -113,11 +109,7 @@ class Product(DeclarativeBase): model_name = model_multi.group(1) base_list = model_multi.group(2) bases = [b.strip() for b in base_list.split(",")] - is_model = any( - b in ("Base", "DeclarativeBase", "DeclarativeMeta") - or b.endswith("Base") - for b in bases - ) + is_model = any(b in ("Base", "DeclarativeBase", "DeclarativeMeta") or b.endswith("Base") for b in bases) if not is_model: i += 1 continue @@ -343,9 +335,7 @@ def _parse_kwargs(self, args: list[str]) -> dict[str, str]: value = kv_match.group(2).strip() # Strip quotes from string values - if (value.startswith('"') and value.endswith('"')) or ( - value.startswith("'") and value.endswith("'") - ): + if (value.startswith('"') and value.endswith('"')) or (value.startswith("'") and value.endswith("'")): kwargs[key] = value[1:-1] elif value.lower() in ("true", "false"): kwargs[key] = value.lower() diff --git a/src/schemaforge/parsers/typeorm_parser.py b/src/schemaforge/parsers/typeorm_parser.py index 4f4bf52..de70907 100644 --- a/src/schemaforge/parsers/typeorm_parser.py +++ b/src/schemaforge/parsers/typeorm_parser.py @@ -92,9 +92,9 @@ def _split_entities(self, text: str) -> list[str]: for line in text.split("\n"): stripped = line.strip() # Detect @Entity, @ViewEntity, or bare class extends pattern - if ( - stripped.startswith("@") or stripped.startswith("export class") - ) and (stripped.startswith("export class") or "class " in stripped): + if (stripped.startswith("@") or stripped.startswith("export class")) and ( + stripped.startswith("export class") or "class " in stripped + ): in_class = True current = line + "\n" brace_depth = 0 @@ -174,18 +174,12 @@ def _parse_entity(self, block: str) -> Table | None: field_line = next_line break - if decorator_text.startswith( - "@PrimaryGeneratedColumn" - ) or decorator_text.startswith("@PrimaryColumn"): - col = self._parse_column_with_decorator( - decorator_text, field_line, is_pk=True - ) + if decorator_text.startswith("@PrimaryGeneratedColumn") or decorator_text.startswith("@PrimaryColumn"): + col = self._parse_column_with_decorator(decorator_text, field_line, is_pk=True) if col: table.columns.append(col) elif decorator_text.startswith("@Column"): - col = self._parse_column_with_decorator( - decorator_text, field_line, is_pk=False - ) + col = self._parse_column_with_decorator(decorator_text, field_line, is_pk=False) if col: table.columns.append(col) elif decorator_text.startswith("@Index"): @@ -197,9 +191,7 @@ def _parse_entity(self, block: str) -> Table | None: return table - def _parse_column_with_decorator( - self, decorator: str, field_line: str, is_pk: bool - ) -> Column | None: + def _parse_column_with_decorator(self, decorator: str, field_line: str, is_pk: bool) -> Column | None: """Parse a @Column or @PrimaryGeneratedColumn decorator + field.""" if not field_line: return None diff --git a/src/schemaforge/type_config.py b/src/schemaforge/type_config.py index ff87d7c..c15433f 100644 --- a/src/schemaforge/type_config.py +++ b/src/schemaforge/type_config.py @@ -46,9 +46,7 @@ def __init__(self, overrides: dict[str, dict[str, str]] | None = None): """ self._overrides: dict[str, dict[str, str]] = overrides or {} - def get_override( - self, col: Column, fmt: str, type_args: dict[str, Any] | None = None - ) -> str | None: + def get_override(self, col: Column, fmt: str, type_args: dict[str, Any] | None = None) -> str | None: """Resolve a custom type override for a column in a format. Args: @@ -74,9 +72,7 @@ def get_override( if args: for arg_key, arg_val in args.items(): if isinstance(arg_val, list): - result = result.replace( - "{" + arg_key + "}", ", ".join(f"'{v}'" for v in arg_val) - ) + result = result.replace("{" + arg_key + "}", ", ".join(f"'{v}'" for v in arg_val)) else: result = result.replace("{" + arg_key + "}", str(arg_val)) # Remove any remaining unresolved placeholders @@ -121,8 +117,7 @@ def from_file(cls, path: str | os.PathLike) -> TypeConfig: raise FileNotFoundError(f"Type config not found: {path}") if path.suffix.lower() not in _CONFIG_EXTENSIONS: raise ValueError( - f"Unsupported config format: {path.suffix}. " - f"Supported: {', '.join(sorted(_CONFIG_EXTENSIONS))}" + f"Unsupported config format: {path.suffix}. Supported: {', '.join(sorted(_CONFIG_EXTENSIONS))}" ) raw: dict[str, Any] = {} @@ -131,8 +126,7 @@ def from_file(cls, path: str | os.PathLike) -> TypeConfig: import yaml # type: ignore[import-untyped] except ImportError: raise ImportError( - "PyYAML is required for YAML type config files. " - "Install it with: pip install pyyaml" + "PyYAML is required for YAML type config files. Install it with: pip install pyyaml" ) from None with open(path) as f: raw = yaml.safe_load(f) or {} diff --git a/tests/test_base.py b/tests/test_base.py index 37c99fc..9763c79 100644 --- a/tests/test_base.py +++ b/tests/test_base.py @@ -61,26 +61,20 @@ def test_build_type_string_simple(): def test_build_type_string_string_with_length(): """build_type_string adds length for STRING.""" col = Column(name="name", type=ColumnType.STRING, type_args={"length": 100}) - result = build_type_string( - col, {ColumnType.STRING: "String"}, string_fmt="{}({})", string_default="String" - ) + result = build_type_string(col, {ColumnType.STRING: "String"}, string_fmt="{}({})", string_default="String") assert result == "String(100)" def test_build_type_string_string_no_length(): """build_type_string returns base for STRING without length.""" col = Column(name="name", type=ColumnType.STRING) - result = build_type_string( - col, {ColumnType.STRING: "String"}, string_fmt="{}({})", string_default="String" - ) + result = build_type_string(col, {ColumnType.STRING: "String"}, string_fmt="{}({})", string_default="String") assert result == "String" def test_build_type_string_decimal(): """build_type_string adds precision/scale for DECIMAL.""" - col = Column( - name="price", type=ColumnType.DECIMAL, type_args={"precision": 12, "scale": 4} - ) + col = Column(name="price", type=ColumnType.DECIMAL, type_args={"precision": 12, "scale": 4}) result = build_type_string( col, {ColumnType.DECIMAL: "DECIMAL"}, @@ -92,9 +86,7 @@ def test_build_type_string_decimal(): def test_build_type_string_enum(): """build_type_string formats inline ENUM values.""" - col = Column( - name="size", type=ColumnType.ENUM, type_args={"values": ["S", "M", "L"]} - ) + col = Column(name="size", type=ColumnType.ENUM, type_args={"values": ["S", "M", "L"]}) result = build_type_string(col, {ColumnType.ENUM: "Enum"}, enum_fmt="Enum({})") assert result == "Enum('S', 'M', 'L')" diff --git a/tests/test_cli.py b/tests/test_cli.py index b5ed620..6d29fea 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -772,6 +772,7 @@ def test_formats_json(self): result = runner.invoke(main, ["formats", "--json"]) assert result.exit_code == 0 import json + parsed = json.loads(result.output.strip()) assert isinstance(parsed, list) assert "sql" in parsed diff --git a/tests/test_diff.py b/tests/test_diff.py index 66021e2..93122ba 100644 --- a/tests/test_diff.py +++ b/tests/test_diff.py @@ -17,9 +17,7 @@ def _col(name: str, col_type: ColumnType = ColumnType.STRING, **kwargs) -> Colum return Column(name=name, type=col_type, **kwargs) -def _table( - name: str, columns: list[Column] | None = None, indexes: list[Index] | None = None -) -> Table: +def _table(name: str, columns: list[Column] | None = None, indexes: list[Index] | None = None) -> Table: return Table(name=name, columns=columns or [], indexes=indexes or []) diff --git a/tests/test_drizzle.py b/tests/test_drizzle.py index 3a16cd7..fc66473 100644 --- a/tests/test_drizzle.py +++ b/tests/test_drizzle.py @@ -136,9 +136,7 @@ def test_generate_simple_table(self): schema = Schema() table = Table(name="users") - table.columns.append( - Column(name="id", type=ColumnType.INTEGER, primary_key=True) - ) + table.columns.append(Column(name="id", type=ColumnType.INTEGER, primary_key=True)) table.columns.append( Column( name="name", @@ -147,9 +145,7 @@ def test_generate_simple_table(self): type_args={"length": 255}, ) ) - table.columns.append( - Column(name="email", type=ColumnType.TEXT, nullable=False, unique=True) - ) + table.columns.append(Column(name="email", type=ColumnType.TEXT, nullable=False, unique=True)) schema.tables.append(table) gen = DrizzleGenerator(dialect="pg") @@ -169,9 +165,7 @@ def test_generate_with_enum(self): schema = Schema() schema.enums.append(EnumType(name="role", values=["admin", "user"])) table = Table(name="users") - table.columns.append( - Column(name="id", type=ColumnType.INTEGER, primary_key=True) - ) + table.columns.append(Column(name="id", type=ColumnType.INTEGER, primary_key=True)) schema.tables.append(table) gen = DrizzleGenerator(dialect="pg") @@ -186,9 +180,7 @@ def test_generate_mysql_dialect(self): schema = Schema() table = Table(name="products") - table.columns.append( - Column(name="id", type=ColumnType.INTEGER, primary_key=True) - ) + table.columns.append(Column(name="id", type=ColumnType.INTEGER, primary_key=True)) schema.tables.append(table) gen = DrizzleGenerator(dialect="mysql") @@ -202,9 +194,7 @@ def test_generate_default_now(self): schema = Schema() table = Table(name="logs") - table.columns.append( - Column(name="createdAt", type=ColumnType.DATETIME, default="now()") - ) + table.columns.append(Column(name="createdAt", type=ColumnType.DATETIME, default="now()")) schema.tables.append(table) gen = DrizzleGenerator(dialect="pg") diff --git a/tests/test_graphql.py b/tests/test_graphql.py index 92e096f..075ef35 100644 --- a/tests/test_graphql.py +++ b/tests/test_graphql.py @@ -238,9 +238,7 @@ def test_generate_simple(self): nullable=False, custom_type="Role", ), - Column( - name="createdAt", type=ColumnType.DATETIME, nullable=False - ), + Column(name="createdAt", type=ColumnType.DATETIME, nullable=False), ], ) ], diff --git a/tests/test_json_schema.py b/tests/test_json_schema.py index 007a3e8..a9cb5b6 100644 --- a/tests/test_json_schema.py +++ b/tests/test_json_schema.py @@ -290,9 +290,7 @@ def test_generate_max_length(): Table( name="Item", columns=[ - Column( - name="name", type=ColumnType.STRING, type_args={"length": 100} - ), + Column(name="name", type=ColumnType.STRING, type_args={"length": 100}), ], ) ] diff --git a/tests/test_mcp_server.py b/tests/test_mcp_server.py index 96add98..1832755 100644 --- a/tests/test_mcp_server.py +++ b/tests/test_mcp_server.py @@ -4,12 +4,16 @@ import pytest import sys +from click.testing import CliRunner from pathlib import Path +from unittest.mock import patch sys.path.insert(0, str(Path(__file__).parent.parent / "src")) pytest.importorskip("mcp", reason="mcp is an optional dependency") +from schemaforge import mcp_server +from schemaforge.cli import main from schemaforge.mcp_server import _FORMATS, create_server @@ -149,3 +153,168 @@ def test_convert_tool_alembic_error(): ) assert "Error" in result assert "generator-only" in result.lower() or "not supported" in result.lower() + + +@pytest.mark.parametrize( + ("args", "transport", "host", "port"), + [ + (["--sse"], "sse", "127.0.0.1", 8000), + (["--sse", "--host", "localhost", "--port", "8765"], "sse", "localhost", 8765), + ([], "stdio", "127.0.0.1", 8000), + (["--host", "localhost", "--port", "8765"], "stdio", "127.0.0.1", 8000), + ], + ids=["default-sse", "custom-sse", "default-stdio", "stdio-ignores-sse-options"], +) +def test_mcp_command_transports(args, transport, host, port): + """Enforce the installed SDK run signature without starting either transport.""" + with patch.object(mcp_server.FastMCP, "run", autospec=True) as run_mock: + result = CliRunner().invoke(mcp_server.mcp_command, args) + + assert result.exit_code == 0, repr(result.exception) + run_mock.assert_called_once() + server = run_mock.call_args.args[0] + assert run_mock.call_args.kwargs == {"transport": transport} + assert server.settings.host == host + assert server.settings.port == port + assert len(server._tool_manager._tools) == 5 + if transport == "sse": + assert f"Starting SchemaForge MCP server on http://{host}:{port}" in result.output + + +def test_create_server_custom_address(): + """Configure the address through the installed SDK constructor.""" + server = create_server(host="localhost", port=8765) + assert server.settings.host == "localhost" + assert server.settings.port == 8765 + + +def test_create_server_without_optional_mcp(monkeypatch): + """Missing MCP keeps the same installation error for both factory calls.""" + monkeypatch.setattr(mcp_server, "FastMCP", None) + with pytest.raises(ImportError, match="The 'mcp' package is required"): + create_server() + with pytest.raises(ImportError, match="The 'mcp' package is required"): + create_server(host="localhost", port=8765) + + +def _link_directory(link, target): + if sys.platform == "win32": + # Junctions exercise canonical link resolution without symlink privileges. + import _winapi + + _winapi.CreateJunction(str(target), str(link)) + else: + link.symlink_to(target, target_is_directory=True) + + +@pytest.fixture(params=["environment", "cwd"]) +def mcp_file_paths(tmp_path, monkeypatch, request): + root = tmp_path / "allowed" + maps = root / "maps" + maps.mkdir(parents=True) + schemas = root / "schemas" + schemas.mkdir() + schema = "CREATE TABLE users (id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL);" + for name in ("a.sql", "b.sql"): + (schemas / name).write_text(schema, encoding="utf-8") + map_path = maps / "types.json" + config = '{"overrides": {"prisma": {"STRING": "String @db.Text"}}}' + map_path.write_text(config, encoding="utf-8") + outside = tmp_path / "allowed-neighbor" + outside.mkdir() + outside_map = outside / "types.json" + outside_map.write_text(config, encoding="utf-8") + _link_directory(root / "linked-outside", outside) + _link_directory(root / "linked-maps", maps) + if request.param == "environment": + monkeypatch.setenv("SCHEMAFORGE_MCP_ROOT", str(root)) + monkeypatch.chdir(tmp_path) + else: + monkeypatch.delenv("SCHEMAFORGE_MCP_ROOT", raising=False) + monkeypatch.chdir(root) + return root, map_path, outside_map, schemas + + +def _invoke_type_map_tool(tool_name, directory, **kwargs): + tool = create_server()._tool_manager._tools[tool_name] + if tool_name == "convert": + return tool.fn("CREATE TABLE users (id INTEGER PRIMARY KEY, name VARCHAR(100) NOT NULL);", **kwargs) + return tool.fn(str(directory), **kwargs) + + +@pytest.mark.parametrize("tool_name", ["convert", "check"]) +@pytest.mark.parametrize("path_kind", ["absolute", "traversal", "symlink"]) +def test_mcp_type_map_rejects_outside_root(mcp_file_paths, tool_name, path_kind): + """Reject outside-root paths before the loader can read any file.""" + root, _, outside_map, schemas = mcp_file_paths + if path_kind == "absolute": + path = outside_map + elif path_kind == "traversal": + path = root.relative_to(Path.cwd()) / ".." / outside_map.parent.name / outside_map.name + else: + path = root / "linked-outside" / outside_map.name + with patch.object(mcp_server.TypeConfig, "from_file", return_value=mcp_server.TypeConfig()) as load: + result = _invoke_type_map_tool(tool_name, schemas, type_map_path=str(path)) + load.assert_not_called() + assert "outside the allowed root" in result + + +@pytest.mark.parametrize("tool_name", ["convert", "check"]) +@pytest.mark.parametrize("path_kind", ["absolute", "relative", "symlink"]) +def test_mcp_type_map_allows_in_root(mcp_file_paths, tool_name, path_kind): + """Load valid synthetic maps through their validated canonical paths.""" + root, map_path, _, schemas = mcp_file_paths + if path_kind == "absolute": + path = map_path + elif path_kind == "relative": + path = map_path.relative_to(Path.cwd()) + else: + path = root / "linked-maps" / map_path.name + with patch.object(mcp_server.TypeConfig, "from_file", wraps=mcp_server.TypeConfig.from_file) as load: + result = _invoke_type_map_tool(tool_name, schemas, type_map_path=str(path)) + load.assert_called_once() + assert Path(load.call_args.args[0]) == map_path.resolve() + if tool_name == "convert": + assert "String @db.Text" in result + else: + assert "PASS: All schema files are equivalent" in result + + +@pytest.mark.parametrize("tool_name", ["convert", "check"]) +def test_mcp_type_map_omitted(mcp_file_paths, tool_name): + """Omitting the map keeps normal conversion and checking without a load.""" + _, _, _, schemas = mcp_file_paths + with patch.object(mcp_server.TypeConfig, "from_file") as load: + result = _invoke_type_map_tool(tool_name, schemas) + load.assert_not_called() + if tool_name == "convert": + assert "model users" in result + else: + assert "PASS: All schema files are equivalent" in result + + +def test_mcp_check_directory_rejects_outside_root(mcp_file_paths): + """The existing check-directory guard still prevents map loading.""" + _, map_path, outside_map, _ = mcp_file_paths + with patch.object(mcp_server.TypeConfig, "from_file") as load: + result = _invoke_type_map_tool("check", outside_map.parent, type_map_path=str(map_path)) + load.assert_not_called() + assert "Directory" in result and "outside the allowed root" in result + + +@pytest.mark.parametrize("tool_name", ["convert", "check"]) +def test_standalone_cli_type_map_unrestricted(mcp_file_paths, tool_name): + """The standalone CLI can still load a user-selected synthetic outside map.""" + _, _, outside_map, schemas = mcp_file_paths + if tool_name == "convert": + args = ["convert", str(schemas / "a.sql"), "--from", "sql", "--to", "prisma"] + else: + args = ["check", "--dir", str(schemas), "--canonical", "prisma"] + with patch.object(mcp_server.TypeConfig, "from_file", wraps=mcp_server.TypeConfig.from_file) as load: + result = CliRunner().invoke(main, [*args, "--type-map", str(outside_map)]) + assert result.exit_code == 0, repr(result.exception) + load.assert_called_once_with(str(outside_map)) + if tool_name == "convert": + assert "String @db.Text" in result.output + else: + assert "PASS: All schema files are equivalent" in result.output diff --git a/tests/test_new_formats.py b/tests/test_new_formats.py index 082a2a1..f2ec6fe 100644 --- a/tests/test_new_formats.py +++ b/tests/test_new_formats.py @@ -522,12 +522,7 @@ def test_sqlalchemy_to_typeorm_roundtrip(): def test_sqlalchemy_to_drizzle_roundtrip(): """Cross-format: SQLAlchemy -> Drizzle.""" drizzle = convert_schema(SIMPLE_SQLALCHEMY, "sqlalchemy", "drizzle") - assert ( - "export" in drizzle - or "pgTable" in drizzle - or "sqliteTable" in drizzle - or "defineTable" in drizzle - ) + assert "export" in drizzle or "pgTable" in drizzle or "sqliteTable" in drizzle or "defineTable" in drizzle assert "users" in drizzle.lower() @@ -709,13 +704,13 @@ def test_alembic_custom_revision(): gen = AlembicGenerator() output = gen.generate( schema, - revision_id="abc123def456", + revision_id="testrevision", down_revision="prev_rev", message="Add items table", ) assert "Add items table" in output - assert "revision = 'abc123def456'" in output + assert "revision = 'testrevision'" in output assert "down_revision = 'prev_rev'" in output diff --git a/tests/test_roundtrip.py b/tests/test_roundtrip.py index aa5ad58..0723a3c 100644 --- a/tests/test_roundtrip.py +++ b/tests/test_roundtrip.py @@ -313,9 +313,7 @@ def test_prisma_generate_complex(): name="Item", columns=[ Column(name="id", type=ColumnType.INTEGER, primary_key=True), - Column( - name="name", type=ColumnType.STRING, type_args={"length": 100} - ), + Column(name="name", type=ColumnType.STRING, type_args={"length": 100}), Column( name="price", type=ColumnType.DECIMAL, @@ -521,9 +519,7 @@ def test_sql_fn_default_generates_without_quotes(): type=ColumnType.DATETIME, default="fn:CURRENT_TIMESTAMP", ), - Column( - name="updated_at", type=ColumnType.DATETIME, default="fn:NOW()" - ), + Column(name="updated_at", type=ColumnType.DATETIME, default="fn:NOW()"), Column( name="token", type=ColumnType.UUID, diff --git a/tests/test_type_config.py b/tests/test_type_config.py index f2023f7..7002505 100644 --- a/tests/test_type_config.py +++ b/tests/test_type_config.py @@ -60,9 +60,7 @@ def test_config_with_length_placeholder(): def test_config_with_precision_scale(): """TypeConfig replaces {precision} and {scale} placeholders.""" config = TypeConfig({"sql": {"DECIMAL": "DECIMAL({precision},{scale})"}}) - col = Column( - name="price", type=ColumnType.DECIMAL, type_args={"precision": 12, "scale": 4} - ) + col = Column(name="price", type=ColumnType.DECIMAL, type_args={"precision": 12, "scale": 4}) result = config.get_override(col, "sql") assert result == "DECIMAL(12,4)" @@ -70,9 +68,7 @@ def test_config_with_precision_scale(): def test_config_with_enum_values(): """TypeConfig replaces {values} placeholder.""" config = TypeConfig({"sql": {"ENUM": "ENUM({values})"}}) - col = Column( - name="size", type=ColumnType.ENUM, type_args={"values": ["S", "M", "L"]} - ) + col = Column(name="size", type=ColumnType.ENUM, type_args={"values": ["S", "M", "L"]}) result = config.get_override(col, "sql") assert result == "ENUM('S', 'M', 'L')" @@ -126,11 +122,7 @@ def test_load_from_yaml_without_pyyaml(): f.write("overrides:\n sql:\n INTEGER: BIGINT\n") tmp_path = f.name try: - _orig_import = ( - __builtins__["__import__"] - if isinstance(__builtins__, dict) - else __builtins__.__import__ - ) + _orig_import = __builtins__["__import__"] if isinstance(__builtins__, dict) else __builtins__.__import__ def _mock_import(name, *args, **kw): if name == "yaml":