schema_sentinel.markdown_utils.markdown

  1import datetime
  2import getpass
  3import logging as log
  4import os
  5
  6import pandas as pd
  7from snakemd import Document
  8from sqlalchemy.sql.elements import and_
  9
 10from ..metadata_manager.model.column import Column
 11from ..metadata_manager.model.comparison import Comparison
 12from ..metadata_manager.model.database import Database
 13from ..metadata_manager.model.function import Function
 14from ..metadata_manager.model.pipe import Pipe
 15from ..metadata_manager.model.procedure import Procedure
 16from ..metadata_manager.model.schema import Schema
 17from ..metadata_manager.model.stage import Stage
 18from ..metadata_manager.model.stream import Stream
 19from ..metadata_manager.model.table import Table
 20from ..metadata_manager.model.table_constraint import TableConstraint
 21from ..metadata_manager.model.task import Task
 22from ..metadata_manager.model.view import View
 23
 24
 25def comparison_to_markdown(src_database: Database, trg_database: Database, session) -> Document:
 26    header_level = 1
 27    header = f"{src_database.__get_name__()} -> {trg_database.__get_name__()})"
 28    log.info(f"Writing {header} comparison report to Markdown")
 29    doc: Document = Document()
 30
 31    # Get author from environment or use system username
 32    author = os.getenv("REPORT_AUTHOR", getpass.getuser())
 33
 34    doc.add_raw(f"""
 35    ---
 36    author: {author}
 37    date:   {datetime.datetime.now().strftime("%a %m %y")}
 38    """)
 39
 40    doc.add_heading(f"{header} comparison report")
 41    doc.add_horizontal_rule()
 42    doc.add_heading("Compared databases", level=header_level)
 43    doc.add_block(src_database.__side_by_side__(trg_database).to_markdown())
 44    header_level += 1
 45
 46    doc.add_heading(
 47        "DB Objects existing in the LEFT database and not present in the RIGHT database", level=header_level
 48    )
 49    comparisons = (
 50        session.query(Comparison)
 51        .filter(
 52            and_(
 53                Comparison.source_database_id == src_database.database_id,
 54                Comparison.target_database_id == trg_database.database_id,
 55                Comparison.comparison_value.like('%"comparison": {"left": "%", "right": null}%'),
 56            )
 57        )
 58        .all()
 59    )
 60    differences = None
 61
 62    for comparison in comparisons:
 63        differences = (
 64            pd.concat([differences, comparison.one_diff()]) if differences is not None else comparison.one_diff()
 65        )
 66
 67    doc.add_block(differences.to_markdown())
 68    differences = None
 69
 70    doc.add_heading("Both sides are different", level=header_level)
 71
 72    comparisons = (
 73        session.query(Comparison)
 74        .filter(
 75            and_(
 76                Comparison.source_database_id == src_database.database_id,
 77                Comparison.target_database_id == trg_database.database_id,
 78                Comparison.comparison_value.not_like('%"comparison": {"left": "%", "right": null}%'),
 79            )
 80        )
 81        .all()
 82    )
 83
 84    for comparison in comparisons:
 85        differences = (
 86            pd.concat([differences, comparison.both_diffs()]) if differences is not None else comparison.both_diffs()
 87        )
 88
 89    doc.add_block(differences.to_markdown())
 90
 91    return doc
 92
 93
 94def db_to_markdown(database: Database, session) -> Document:
 95    header = f"{database.__get_name__()}"
 96    log.info(f"Writing {header} database to Markdown")
 97    doc: Document = Document()
 98
 99    # Get author from environment or use system username
100    author = os.getenv("REPORT_AUTHOR", getpass.getuser())
101
102    doc.add_raw(f"""
103    ---
104    author: {author}
105    date:   {datetime.datetime.now().strftime("%a %m %y")}
106    """)
107
108    doc.add_heading(f"{header} database documentation")
109    doc.add_horizontal_rule()
110    doc.add_block(database.__get_df__().to_markdown())
111
112    schemas = session.query(Schema).filter(and_(Schema.database_id == database.database_id)).all()
113    header_level = 1
114    doc.add_heading(f"{database.database_name} schemas", level=header_level)
115    doc.add_block(
116        Schema.__to_df__(schemas, columns=["schema_name", "created", "last_altered", "comment"]).to_markdown()
117    )
118
119    header_level = 2
120    for schema in schemas:
121        schema_md = schema.schema_name.replace("_", "\\_")
122        doc.add_heading(f"Schema: {schema.schema_name}", level=header_level)
123        doc.add_block(schema.__get_df__().to_markdown())
124
125        tables = session.query(Table).filter(Table.schema_id == schema.schema_id).all()
126        if tables is not None:
127            header_level += 1
128            df = Table.__to_df__(tables, ["table_name", "created", "last_altered", "comment"])
129            if df.size:
130                doc.add_heading(f"{schema_md} Tables", level=header_level)
131                doc.add_block(df.to_markdown())
132            header_level -= 1
133
134            header_level += 1
135            for table in tables:
136                table_md = table.table_name.replace("_", "\\_")
137                doc.add_heading(f"Table  {schema.schema_name}.{table_md} ", level=header_level)
138                doc.add_block(table.__get_df__().to_markdown())
139
140                get_object_doc(
141                    data=session.query(Column).filter(Column.table_id == table.table_id).all(),
142                    klass=Column,
143                    columns=[
144                        "column_name",
145                        "ordinal_position",
146                        "is_nullable",
147                        "character_maximum_length",
148                        "numeric_precision",
149                        "numeric_scale",
150                        "datetime_precision",
151                    ],
152                    header=f"Table {schema.schema_name}.{table_md} columns",
153                    doc=doc,
154                    header_level=header_level,
155                )
156
157                get_object_doc(
158                    data=session.query(TableConstraint).filter(TableConstraint.table_id == table.table_id).all(),
159                    klass=TableConstraint,
160                    columns=["table_constraint_name", "constraint_type", "is_deferrable", "created", "last_altered"],
161                    header=f"Table {schema.schema_name}.{table_md} constraints",
162                    doc=doc,
163                    header_level=header_level,
164                )
165
166            header_level -= 1
167
168        get_object_doc(
169            data=session.query(View).filter(View.schema_id == schema.schema_id).all(),
170            klass=View,
171            columns=["view_name", "created", "is_secure", "is_materialized", "enable_schema_evolution", "comment"],
172            header=f"{schema_md} Views",
173            doc=doc,
174            header_level=header_level,
175        )
176
177        get_object_doc(
178            data=session.query(Procedure).filter(Procedure.schema_id == schema.schema_id).all(),
179            klass=Procedure,
180            columns=["procedure_name", "data_type", "argument_signature", "created", "last_altered", "comment"],
181            header=f"{schema_md} Procedures",
182            doc=doc,
183            header_level=header_level,
184        )
185
186        get_object_doc(
187            data=session.query(Function).filter(Function.schema_id == schema.schema_id).all(),
188            klass=Function,
189            columns=["function_name", "data_type", "argument_signature", "created", "last_altered", "comment"],
190            header=f"{schema_md} Functions",
191            doc=doc,
192            header_level=header_level,
193        )
194
195        get_object_doc(
196            data=session.query(Stage).filter(Stage.schema_id == schema.schema_id).all(),
197            klass=Stage,
198            columns=["stage_name", "stage_url", "stage_type", "storage_integration", "created", "comment"],
199            header=f"{schema_md} Stages",
200            doc=doc,
201            header_level=header_level,
202        )
203
204        get_object_doc(
205            data=session.query(Pipe).filter(Pipe.schema_id == schema.schema_id).all(),
206            klass=Pipe,
207            columns=["pipe_name", "pipe_definition", "notification_channel_name", "pattern", "created", "last_altered"],
208            header=f"{schema_md} Pipes",
209            doc=doc,
210            header_level=header_level,
211        )
212
213        get_object_doc(
214            data=session.query(Stream).filter(Stream.schema_id == schema.schema_id).all(),
215            klass=Stream,
216            columns=[
217                "stream_name",
218                "table_name",
219                "source_type",
220                "base_tables",
221                "type",
222                "stale",
223                "invalid_reason",
224                "created",
225                "comment",
226            ],
227            header=f"{schema_md} Streams",
228            doc=doc,
229            header_level=header_level,
230        )
231
232        get_object_doc(
233            data=session.query(Task).filter(Task.schema_id == schema.schema_id).all(),
234            klass=Stream,
235            columns=[
236                "id",
237                "task_name",
238                "warehouse",
239                "schedule",
240                "state",
241                "condition",
242                "error_integration",
243                "config",
244                "last_committed",
245                "last_suspended",
246                "created",
247                "comment",
248            ],
249            header=f"{schema_md} Streams",
250            doc=doc,
251            header_level=header_level,
252        )
253
254    return doc
255
256
257def get_object_doc(data, klass, columns: list, header: str, doc: Document, header_level: int = 0):
258    header_level += 1
259    if data:
260        doc.add_heading(header, level=header_level)
261        df: pd.DataFrame = klass.__to_df__(data, columns)
262        if df.size:
263            doc.add_block(df.to_markdown())
264    header_level -= 1