def ensure_tables_exist()

in antlir/rpm/repo_db.py [0:0]

• 21 lines of code
• 9 McCabe index (conditional complexity)

    def ensure_tables_exist(self, _ensure_line_is_covered=lambda: None):
        # Future: it would be better if this function checked that the table
        # schemas in the DB are exactly as we would create them, and that
        # the DB's "canonical hash" algorithm matches ours.  For now, we'll
        # just trust that future developers will be careful to migrate the
        # DB correctly.
        with self._cursor() as cursor:
            col_types = self._DIALECT_TO_COLUMN_TYPE[self._dialect]
            for table, cols in self._TABLE_COLUMNS.items():
                # On MySQL, `IF NOT EXISTS` raises a warning, so `try` instead.
                try:
                    cursor.execute(
                        f"""
                        CREATE TABLE `{table}` (
                            {', '.join([
                                f'`{c}` {col_types[c]} {d}'
                                    for c, d in cols.items()
                            ] + list(self._TABLE_KEYS[table]))}
                        );
                    """
                    )
                except Exception as ex:
                    # The intent is to catch the 'aready exists' variants of
                    # {sqlite3,_mysql_exceptions}.OperationalError.  But, I
                    # don't want to import MySQLdb here, since it is an
                    # optional dependency for this specific module.
                    if (
                        type(ex).__qualname__ != "OperationalError"
                        or type(ex).__module__
                        not in [
                            "sqlite3",
                            # MySQLdb versions vary the module path.
                            "_mysql_exceptions",
                            "MySQLdb._exceptions",
                        ]
                        or "already exists" not in str(ex.args)
                    ):
                        raise  # pragma: no cover
                    _ensure_line_is_covered()