Plain is headed towards 1.0! Subscribe for development updates →

  1from collections import namedtuple
  2
  3# Structure returned by DatabaseIntrospection.get_table_list()
  4TableInfo = namedtuple("TableInfo", ["name", "type"])
  5
  6# Structure returned by the DB-API cursor.description interface (PEP 249)
  7FieldInfo = namedtuple(
  8    "FieldInfo",
  9    "name type_code display_size internal_size precision scale null_ok "
 10    "default collation",
 11)
 12
 13
 14class BaseDatabaseIntrospection:
 15    """Encapsulate backend-specific introspection utilities."""
 16
 17    data_types_reverse = {}
 18
 19    def __init__(self, connection):
 20        self.connection = connection
 21
 22    def get_field_type(self, data_type, description):
 23        """
 24        Hook for a database backend to use the cursor description to
 25        match a Plain field type to a database column.
 26
 27        For Oracle, the column data_type on its own is insufficient to
 28        distinguish between a FloatField and IntegerField, for example.
 29        """
 30        return self.data_types_reverse[data_type]
 31
 32    def identifier_converter(self, name):
 33        """
 34        Apply a conversion to the identifier for the purposes of comparison.
 35
 36        The default identifier converter is for case sensitive comparison.
 37        """
 38        return name
 39
 40    def table_names(self, cursor=None, include_views=False):
 41        """
 42        Return a list of names of all tables that exist in the database.
 43        Sort the returned table list by Python's default sorting. Do NOT use
 44        the database's ORDER BY here to avoid subtle differences in sorting
 45        order between databases.
 46        """
 47
 48        def get_names(cursor):
 49            return sorted(
 50                ti.name
 51                for ti in self.get_table_list(cursor)
 52                if include_views or ti.type == "t"
 53            )
 54
 55        if cursor is None:
 56            with self.connection.cursor() as cursor:
 57                return get_names(cursor)
 58        return get_names(cursor)
 59
 60    def get_table_list(self, cursor):
 61        """
 62        Return an unsorted list of TableInfo named tuples of all tables and
 63        views that exist in the database.
 64        """
 65        raise NotImplementedError(
 66            "subclasses of BaseDatabaseIntrospection may require a get_table_list() "
 67            "method"
 68        )
 69
 70    def get_table_description(self, cursor, table_name):
 71        """
 72        Return a description of the table with the DB-API cursor.description
 73        interface.
 74        """
 75        raise NotImplementedError(
 76            "subclasses of BaseDatabaseIntrospection may require a "
 77            "get_table_description() method."
 78        )
 79
 80    def get_migratable_models(self):
 81        from plain.models import models_registry
 82        from plain.packages import packages_registry
 83
 84        return (
 85            model
 86            for package_config in packages_registry.get_package_configs()
 87            for model in models_registry.get_models(
 88                package_label=package_config.package_label
 89            )
 90            if model._meta.can_migrate(self.connection)
 91        )
 92
 93    def plain_table_names(self, only_existing=False, include_views=True):
 94        """
 95        Return a list of all table names that have associated Plain models and
 96        are in INSTALLED_PACKAGES.
 97
 98        If only_existing is True, include only the tables in the database.
 99        """
100        tables = set()
101        for model in self.get_migratable_models():
102            tables.add(model._meta.db_table)
103            tables.update(f.m2m_db_table() for f in model._meta.local_many_to_many)
104        tables = list(tables)
105        if only_existing:
106            existing_tables = set(self.table_names(include_views=include_views))
107            tables = [
108                t for t in tables if self.identifier_converter(t) in existing_tables
109            ]
110        return tables
111
112    def sequence_list(self):
113        """
114        Return a list of information about all DB sequences for all models in
115        all packages.
116        """
117        sequence_list = []
118        with self.connection.cursor() as cursor:
119            for model in self.get_migratable_models():
120                sequence_list.extend(
121                    self.get_sequences(
122                        cursor, model._meta.db_table, model._meta.local_fields
123                    )
124                )
125        return sequence_list
126
127    def get_sequences(self, cursor, table_name, table_fields=()):
128        """
129        Return a list of introspected sequences for table_name. Each sequence
130        is a dict: {'table': <table_name>, 'column': <column_name>}. An optional
131        'name' key can be added if the backend supports named sequences.
132        """
133        raise NotImplementedError(
134            "subclasses of BaseDatabaseIntrospection may require a get_sequences() "
135            "method"
136        )
137
138    def get_relations(self, cursor, table_name):
139        """
140        Return a dictionary of {field_name: (field_name_other_table, other_table)}
141        representing all foreign keys in the given table.
142        """
143        raise NotImplementedError(
144            "subclasses of BaseDatabaseIntrospection may require a "
145            "get_relations() method."
146        )
147
148    def get_primary_key_column(self, cursor, table_name):
149        """
150        Return the name of the primary key column for the given table.
151        """
152        columns = self.get_primary_key_columns(cursor, table_name)
153        return columns[0] if columns else None
154
155    def get_primary_key_columns(self, cursor, table_name):
156        """Return a list of primary key columns for the given table."""
157        for constraint in self.get_constraints(cursor, table_name).values():
158            if constraint["primary_key"]:
159                return constraint["columns"]
160        return None
161
162    def get_constraints(self, cursor, table_name):
163        """
164        Retrieve any constraints or keys (unique, pk, fk, check, index)
165        across one or more columns.
166
167        Return a dict mapping constraint names to their attributes,
168        where attributes is a dict with keys:
169         * columns: List of columns this covers
170         * primary_key: True if primary key, False otherwise
171         * unique: True if this is a unique constraint, False otherwise
172         * foreign_key: (table, column) of target, or None
173         * check: True if check constraint, False otherwise
174         * index: True if index, False otherwise.
175         * orders: The order (ASC/DESC) defined for the columns of indexes
176         * type: The type of the index (btree, hash, etc.)
177
178        Some backends may return special constraint names that don't exist
179        if they don't name constraints of a certain type (e.g. SQLite)
180        """
181        raise NotImplementedError(
182            "subclasses of BaseDatabaseIntrospection may require a get_constraints() "
183            "method"
184        )