Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions db2sql/domain/model/column.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ class Column:
computed_definition: Optional[str] = None
identity: bool = False
constraint: Optional[str] = None
comment: Optional[str] = None

@property
def is_primary_key(self) -> bool:
Expand Down
1 change: 1 addition & 0 deletions db2sql/domain/model/schema.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ class Schema:

name: str
tables: Dict[str, Table] = field(default_factory=dict)
comment: Optional[str] = None

def add_table(self, table: Table) -> None:
if table.name in self.tables:
Expand Down
1 change: 1 addition & 0 deletions db2sql/domain/model/table.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ class Table:
indexes: Dict[str, List[str]] = field(default_factory=dict)
foreign_keys: List[ForeignKey] = field(default_factory=list)
source_query: Optional[str] = None
comment: Optional[str] = None

def add_column(self, column: Column) -> None:
if column.name in self.columns:
Expand Down
71 changes: 66 additions & 5 deletions db2sql/infrastructure/emit/mssql/emitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

import re
import warnings
from typing import Any, Dict, Iterable, Mapping, Optional
from typing import Any, Dict, Iterable, Mapping, Optional, Tuple

from db2sql.application.ports import OutputSink
from db2sql.domain.model import Column, Database, Schema, Table
Expand Down Expand Up @@ -246,13 +246,34 @@ def emit_epilogue(self, sink: OutputSink) -> None:
sink.write("COMMIT TRANSACTION;\n")
sink.boundary()

@staticmethod
def _string_literal(value: str) -> str:
escaped = value.replace("'", "''")
return f"N'{escaped}'"

def _description_arguments(self, comment: str, levels: Iterable[Tuple[str, str]]) -> str:
"""Arguments of ``sp_addextendedproperty`` for one ``MS_Description``.

The level names are the identifiers as created, not quoted: the
procedure takes them as plain strings.
"""
arguments = [
"@name = N'MS_Description'",
f"@value = {self._string_literal(comment)}",
]
for index, (level_type, level_name) in enumerate(levels):
arguments.append(f"@level{index}type = N'{level_type}'")
arguments.append(f"@level{index}name = {self._string_literal(level_name)}")
return ", ".join(arguments)

def emit_schemas(self, database: Database, sink: OutputSink) -> None:
emitted = set()
targets: Dict[str, Optional[str]] = {}
for schema in database.schemas.values():
target = self._schema_mapping.get(schema.name, schema.name)
if target in emitted:
continue
emitted.add(target)
# Several source schemas can map to one target: the first comment wins.
if targets.get(target) is None:
targets[target] = schema.comment
for target, comment in targets.items():
quoted = self.quote_identifier(target)
# CREATE SCHEMA must be the first statement in its batch — wrap in EXEC.
literal = target.replace("'", "''")
Expand All @@ -261,8 +282,24 @@ def emit_schemas(self, database: Database, sink: OutputSink) -> None:
f" EXEC('CREATE SCHEMA {quoted}');\n"
)
sink.boundary()
if comment is not None:
self._emit_schema_comment(target, comment, sink)
sink.write("\n")

def _emit_schema_comment(self, target: str, comment: str, sink: OutputSink) -> None:
# The schema can exist before the script runs (``dbo`` always does) and
# already carry a description: sp_addextendedproperty then fails.
name = self._string_literal(self._normalize(target))
arguments = self._description_arguments(comment, [("SCHEMA", self._normalize(target))])
sink.write(
"IF EXISTS (SELECT 1 FROM sys.fn_listextendedproperty("
f"N'MS_Description', N'SCHEMA', {name}, NULL, NULL, NULL, NULL))\n"
f" EXEC sys.sp_updateextendedproperty {arguments};\n"
"ELSE\n"
f" EXEC sys.sp_addextendedproperty {arguments};\n"
)
sink.boundary()

def emit_drops(self, database: Database, sink: OutputSink) -> None:
# Native ``DROP TABLE IF EXISTS`` syntax — requires SQL Server 2016+.
for schema_name, table_name in drop_order(database):
Expand Down Expand Up @@ -309,6 +346,30 @@ def emit_tables(self, database: Database, sink: OutputSink) -> None:
sink.write(",\n".join(column_lines))
sink.write("\n);\n\n")
sink.boundary()
self._emit_table_comments(schema, table, sink)

def _emit_table_comments(self, schema: Schema, table: Table, sink: OutputSink) -> None:
# The table is created by this script, so it has no description yet.
levels = [
("SCHEMA", self._normalize(self._schema_mapping.get(schema.name, schema.name))),
("TABLE", self._normalize(table.name)),
]
emitted = False
if table.comment is not None:
arguments = self._description_arguments(table.comment, levels)
sink.write(f"EXEC sys.sp_addextendedproperty {arguments};\n")
sink.boundary()
emitted = True
for column in table.columns.values():
if column.comment is None:
continue
column_levels = levels + [("COLUMN", self._normalize(column.name))]
arguments = self._description_arguments(column.comment, column_levels)
sink.write(f"EXEC sys.sp_addextendedproperty {arguments};\n")
sink.boundary()
emitted = True
if emitted:
sink.write("\n")

def emit_foreign_keys(self, database: Database, sink: OutputSink) -> None:
for schema in database.schemas.values():
Expand Down
35 changes: 30 additions & 5 deletions db2sql/infrastructure/emit/postgres/emitter.py
Original file line number Diff line number Diff line change
Expand Up @@ -359,15 +359,29 @@ def emit_epilogue(self, sink: OutputSink) -> None:
sink.write("COMMIT;\n")
sink.boundary()

@staticmethod
def _string_literal(value: str) -> str:
escaped = value.replace("'", "''")
return f"'{escaped}'"

def _emit_comment(self, sink: OutputSink, target: str, comment: Optional[str]) -> None:
if comment is None:
return
sink.write(f"COMMENT ON {target} IS {self._string_literal(comment)};\n")
sink.boundary()

def emit_schemas(self, database: Database, sink: OutputSink) -> None:
emitted = set()
targets: Dict[str, Optional[str]] = {}
for schema in database.schemas.values():
target = self._schema_mapping.get(schema.name, schema.name)
if target in emitted:
continue
emitted.add(target)
sink.write(f"CREATE SCHEMA IF NOT EXISTS {self.quote_identifier(target)};\n")
# Several source schemas can map to one target: the first comment wins.
if targets.get(target) is None:
targets[target] = schema.comment
for target, comment in targets.items():
quoted = self.quote_identifier(target)
sink.write(f"CREATE SCHEMA IF NOT EXISTS {quoted};\n")
sink.boundary()
self._emit_comment(sink, f"SCHEMA {quoted}", comment)
sink.write("\n")

def emit_drops(self, database: Database, sink: OutputSink) -> None:
Expand Down Expand Up @@ -414,6 +428,17 @@ def emit_tables(self, database: Database, sink: OutputSink) -> None:
sink.write(",\n".join(column_lines))
sink.write("\n);\n\n")
sink.boundary()
self._emit_table_comments(schema, table, sink)

def _emit_table_comments(self, schema: Schema, table: Table, sink: OutputSink) -> None:
qualified = self.table_name(schema, table)
self._emit_comment(sink, f"TABLE {qualified}", table.comment)
commented = [col for col in table.columns.values() if col.comment is not None]
for column in commented:
target = f"COLUMN {qualified}.{self.quote_identifier(column.name)}"
self._emit_comment(sink, target, column.comment)
if table.comment is not None or commented:
sink.write("\n")

def emit_foreign_keys(self, database: Database, sink: OutputSink) -> None:
for schema in database.schemas.values():
Expand Down
47 changes: 47 additions & 0 deletions db2sql/infrastructure/persistence/mssql/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,8 @@ def collect_metadata(self) -> Database:
self._read_foreign_keys(database)
self._logger.info("reading indexes")
self._read_indexes(database)
self._logger.info("reading comments")
self._read_comments(database)
except Exception as exception:
self._logger.error(str(exception))
raise SourceReaderError("failed to collect database information") from exception
Expand Down Expand Up @@ -276,3 +278,48 @@ def _read_indexes(self, database: Database) -> None:
table = database.get_table(row.TABLE_SCHEMA, row.TABLE_NAME)
if table:
table.add_index(row.INDEX_NAME, row.COLUMN_NAME)

def _read_comments(self, database: Database) -> None:
# SQL Server has no COMMENT ON. SSMS and most tools store the comment
# as the MS_Description extended property: class 3 on a schema, class 1
# on a table (minor_id 0) or on one of its columns (minor_id = column_id).
r: engine.Result[Any] = self._ensure_session().execute(text("""
SELECT s.name AS TABLE_SCHEMA,
NULL AS TABLE_NAME,
NULL AS COLUMN_NAME,
CAST(ep.value AS NVARCHAR(MAX)) AS COMMENT
FROM sys.extended_properties ep
INNER JOIN sys.schemas s
ON ep.major_id = s.schema_id
WHERE ep.class = 3
AND ep.name = 'MS_Description'
UNION ALL
SELECT s.name, t.name, c.name, CAST(ep.value AS NVARCHAR(MAX))
FROM sys.extended_properties ep
INNER JOIN sys.tables t
ON ep.major_id = t.object_id
INNER JOIN sys.schemas s
ON s.schema_id = t.schema_id
LEFT JOIN sys.columns c
ON c.object_id = t.object_id
AND c.column_id = ep.minor_id
WHERE ep.class = 1
AND ep.name = 'MS_Description'
AND (ep.minor_id = 0 OR c.column_id IS NOT NULL)
"""))

for row in r:
if row.TABLE_NAME is None:
schema = database.get_schema(row.TABLE_SCHEMA)
if schema is not None:
schema.comment = row.COMMENT
continue
table = database.get_table(row.TABLE_SCHEMA, row.TABLE_NAME)
if table is None:
continue
if row.COLUMN_NAME is None:
table.comment = row.COMMENT
continue
column = table.get_column(row.COLUMN_NAME)
if column is not None:
column.comment = row.COMMENT
11 changes: 8 additions & 3 deletions db2sql/infrastructure/persistence/mysql/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,21 +69,24 @@ def collect_metadata(self) -> Database:
def _read_tables(self, database: Database) -> None:
rows = self._ensure_session().execute(
text(
"SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES "
"SELECT TABLE_NAME, TABLE_COMMENT FROM INFORMATION_SCHEMA.TABLES "
"WHERE TABLE_SCHEMA = :schema AND TABLE_TYPE = 'BASE TABLE' "
"ORDER BY TABLE_NAME"
),
{"schema": self._database_name},
)
for row in rows:
database.add_table(self._database_name, Table(row.TABLE_NAME))
database.add_table(
self._database_name,
Table(row.TABLE_NAME, comment=row.TABLE_COMMENT or None),
)

def _read_columns(self, database: Database) -> None:
rows = self._ensure_session().execute(
text(
"SELECT TABLE_NAME, COLUMN_NAME, COLUMN_DEFAULT, IS_NULLABLE, "
"DATA_TYPE, CHARACTER_MAXIMUM_LENGTH, NUMERIC_PRECISION, "
"NUMERIC_SCALE, EXTRA "
"NUMERIC_SCALE, EXTRA, COLUMN_COMMENT "
"FROM INFORMATION_SCHEMA.COLUMNS "
"WHERE TABLE_SCHEMA = :schema "
"ORDER BY TABLE_NAME, ORDINAL_POSITION"
Expand All @@ -102,6 +105,8 @@ def _read_columns(self, database: Database) -> None:
char_length=row.CHARACTER_MAXIMUM_LENGTH or -1,
precision=row.NUMERIC_PRECISION,
scale=row.NUMERIC_SCALE,
# MySQL reports "no comment" as an empty string, not NULL.
comment=row.COLUMN_COMMENT or None,
)
if row.EXTRA and "auto_increment" in row.EXTRA.lower():
column.identity = True
Expand Down
41 changes: 41 additions & 0 deletions db2sql/infrastructure/persistence/oracle/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -139,6 +139,8 @@ def collect_metadata(self) -> Database:
self._read_indexes(database)
self._logger.info("reading identity columns ...")
self._read_identity_columns(database)
self._logger.info("reading comments ...")
self._read_comments(database)
except SourceReaderError:
raise
except Exception as exc:
Expand Down Expand Up @@ -361,6 +363,45 @@ def _read_identity_columns(self, database: Database) -> None:
if column is not None:
column.identity = True

def _read_comments(self, database: Database) -> None:
"""Oracle has no comment on a schema, only on tables and columns."""
owner = self._schema_filter
params: Dict[str, Any] = {}
if owner:
owner_clause = "OWNER = :owner"
params["owner"] = owner
else:
owner_clause = self._excluded_schemas_sql("OWNER")
session = self._ensure_session()
tables = session.execute(
text(
"SELECT OWNER, TABLE_NAME, COMMENTS "
"FROM ALL_TAB_COMMENTS "
f"WHERE {owner_clause} "
" AND TABLE_TYPE = 'TABLE' AND COMMENTS IS NOT NULL"
),
params,
)
for row in tables:
table = database.get_table(row.owner, row.table_name)
if table is not None:
table.comment = row.comments
columns = session.execute(
text(
"SELECT OWNER, TABLE_NAME, COLUMN_NAME, COMMENTS "
"FROM ALL_COL_COMMENTS "
f"WHERE {owner_clause} AND COMMENTS IS NOT NULL"
),
params,
)
for row in columns:
table = database.get_table(row.owner, row.table_name)
if table is None:
continue
column = table.get_column(row.column_name)
if column is not None:
column.comment = row.comments

def iter_rows(self, schema: str, table: Table, limit: int = -1) -> Iterator[Tuple[Any, ...]]:
session = self._ensure_session()
columns = ", ".join(f'"{name}"' for name in table.columns)
Expand Down
45 changes: 45 additions & 0 deletions db2sql/infrastructure/persistence/postgres/reader.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
from db2sql.infrastructure.url import build_url, redact_url

_SYSTEM_SCHEMAS = ("pg_catalog", "information_schema", "pg_toast")
_PUBLIC_SCHEMA_DEFAULT_COMMENT = ("public", "standard public schema")


class PostgresSourceReader:
Expand Down Expand Up @@ -50,6 +51,7 @@ def collect_metadata(self) -> Database:
self._read_constraints(database)
self._read_foreign_keys(database)
self._read_indexes(database)
self._read_comments(database)
except Exception as exc:
self._logger.error(str(exc))
raise SourceReaderError("failed to collect database information") from exc
Expand Down Expand Up @@ -175,6 +177,49 @@ def _read_indexes(self, database: Database) -> None:
if table:
table.add_index(row.index_name, row.column_name)

def _read_comments(self, database: Database) -> None:
# One row per COMMENT ON: objsubid 0 is the object itself, a positive
# objsubid is the attnum of a column of that table.
rows = self._ensure_session().execute(
text(
"SELECT n.nspname AS schema_name, NULL AS table_name, "
" 0 AS column_number, NULL AS column_name, d.description "
"FROM pg_description d "
"JOIN pg_namespace n ON n.oid = d.objoid "
"WHERE d.classoid = 'pg_namespace'::regclass "
f" AND n.nspname NOT IN {_SYSTEM_SCHEMAS} "
"UNION ALL "
"SELECT n.nspname, c.relname, d.objsubid, a.attname, d.description "
"FROM pg_description d "
"JOIN pg_class c ON c.oid = d.objoid "
"JOIN pg_namespace n ON n.oid = c.relnamespace "
"LEFT JOIN pg_attribute a "
" ON a.attrelid = c.oid AND a.attnum = d.objsubid "
"WHERE d.classoid = 'pg_class'::regclass "
" AND c.relkind IN ('r', 'p') "
f" AND n.nspname NOT IN {_SYSTEM_SCHEMAS}"
)
)
for row in rows:
if row.table_name is None:
schema = database.get_schema(row.schema_name)
# PG ships ``public`` with this comment. Copying it would make
# the dump COMMENT ON a schema that the target role usually
# does not own, which fails the whole transaction.
builtin = (row.schema_name, row.description) == _PUBLIC_SCHEMA_DEFAULT_COMMENT
if schema is not None and not builtin:
schema.comment = row.description
continue
table = database.get_table(row.schema_name, row.table_name)
if table is None:
continue
if row.column_number == 0:
table.comment = row.description
continue
column = table.get_column(row.column_name) if row.column_name else None
if column is not None:
column.comment = row.description

def iter_rows(self, schema: str, table: Table, limit: int = -1) -> Iterator[Tuple[Any, ...]]:
session = self._ensure_session()
columns = ", ".join(f'"{name}"' for name in table.columns)
Expand Down
Loading
Loading