Source code for dbschema.objects

# -*- coding: utf-8 -*-

import importlib

from ..exceptions import DBSchemaError


DOESNOTEXIST = object()


[docs]class Node(object): """Base class for database objects.""" def __init__(self, name, **kwargs): """Constructor.""" # TODO(andi) maybe it's more efficient to store children in a # dict (oid -> object)? It would improve oid lookups and it # would be much easier to replace a child object. self.children = set([]) self.parent = None self.name = name #: The name of the object. self.description = None #: The description of the object. self.oid = None #: A unique identifier provided by the underlying backend. for key in kwargs: setattr(self, key, kwargs[key]) def __repr__(self): return '<%s:%s at 0x%0x>' % (self.__class__.__name__, self.name or '??', id(self)) def __hash__(self): return hash((self.__class__, self.parent, self.name)) def __eq__(self, other): return hash(self) == hash(other) def __lt__(self, other): return self.name < other.name def _print_tree(self, level=0): # NOQA # use for debugging, prints the tree print('%s%r' % (' ' * level, self)) for child in self.children: child._print_tree(level + 1) @property def db(self): if isinstance(self, Database): return self return self.parent.db
[docs] def add_child(self, obj): """Adds a child objects.""" obj.parent = self if obj not in self.children: self.children.add(obj) # TODO(andi): This assumes that the node is already child of a root # Database node which makes it impossible to create a sub-tree that # should be added to the real root later. For example: # db = Database() # node = Node() # db.add_child(node) # sub = Node() # sub.add_child(Node()) # <-- fails # node.add_child(db) self.db._oid_idx[obj.oid] = obj return obj
[docs] def find(self, type_cls=None, name=None, parent=None, recurse=True, **kwargs): """Yields database objects matching search parameters. All search parameters are optional. If not parameters are given all database objects are returned. :param type_cls: Find specific types. :type type_cls: Subclass of :class:`Node` :param name: The name to match. :type name: str :param parent: The parent of the objects that should be yieled. :type parent: Instance of :class:`Node` :param kwargs: All other keyword parameters are used to compare with the attributes of the child instances. :param recurse: If ``True`` (the default) recurse into children. :returns: Iterator of :class:`Node` instances. """ for child in self.children: yield_child = True if type_cls is not None and not isinstance(child, type_cls): yield_child = False if name is not None and child.name != name: yield_child = False if parent is not None and child.parent != parent: yield_child = False for key in kwargs: child_value = getattr(child, key, DOESNOTEXIST) if child_value == DOESNOTEXIST or child_value != kwargs[key]: yield_child = False break if yield_child: yield child if recurse: # yield from FTW! But we want to support older Python # versions too... for cchild in child.find(type_cls=type_cls, name=name, parent=parent, recurse=recurse, **kwargs): yield cchild
[docs] def find_exact(self, **kwargs): """Returns exact one match or None. For parameter reference see :func:`find`. """ results = list(self.find(**kwargs)) if len(results) == 1: return results[0] return None
[docs] def get_child_types(self): """Returns a set of child types for this node.""" types = set() for child in self.children: types.add(child.__class__) return types
[docs]class Database(Node): """Central database class.""" dbapi_module = None structure = [] def __init__(self, name): super(Database, self).__init__(name) self._conn = None self._conn_kwargs = None self._oid_idx = {} self._dirty = set() for type_cls, children in self.structure: self._populate_dirty(type_cls, children) def _populate_dirty(self, type_cls, children): self._dirty.add(type_cls) children = children or [] for type_cls, cchildren in children: self._populate_dirty(type_cls, cchildren)
[docs] def set_connection(self, connection, **connect_kwargs): """Sets the connection to interact with the database. :param connection: If not ``None``, a already opened database connection. :type connection: DB-API2 connection or None :param connect_kwargs: Connection kwargs if not connection is given. It is an error if both ``connection`` and ``connect_kwargs`` are given. """ if connection and connect_kwargs: raise DBSchemaError( 'Either connection or connect_kwargs is allowed, not both.') elif connection: self._conn = connection self._conn_kwargs = None else: self._conn = None self._conn_kwargs = connect_kwargs
@property def dbapi(self): try: dbapi = importlib.import_module(self.dbapi_module) except ImportError as err: raise DBSchemaError( 'Failed to load database module: %s' % err) return dbapi @property def connection(self): if self._conn is None: try: self._conn = self.dbapi.connect(**self._conn_kwargs) except Exception as err: raise DBSchemaError( 'Could not connect to database: %s' % err) return self._conn def _initialize(self): """Initializes database objects. This method is called after the creation of a :class:`Database` instance. It's purpose is to pre-populate the tree of database objects. Backends don't need to implement this method, but if they do it should perform a quick operation to fetch basic information about the database. """ pass def find_by_oid(self, oid): return self._oid_idx.get(oid, None)
[docs] def get_server_info(self): """Returns version information of the database. :rtype: str """ raise NotImplementedError('Database.get_version()')
[docs] def get_default_namespace(self): """Returns the default namespace or None. :rtype: :class:`Namespace` or ``None`` """ return None
def _get_types_from_default_ns(self, type_cls): """Helper to yield objects from default namespace.""" self._refresh_types_internal([type_cls]) nsp = self.get_default_namespace() or self return nsp.find(type_cls=type_cls)
[docs] def get_tables(self): """Yields all tables from default namespace. :rtype: Generator of :class:`Table` instances. """ return self._get_types_from_default_ns(Table)
[docs] def get_views(self): """Yields all views from the default namespace. :rtype: Generator of :class:`View` instances- """ return self._get_types_from_default_ns(View)
# def refresh(self, obj, type_cls=None): # """Updates obj. # If type_cls is given, the given object is requested to be # updated in regard to the given type_cls. For example, if obj # is a table and type_cls :class:`Column` then this function # should populate the columns for obj. # :param type_cls: A :class:`Node` subclass. # """ # pass
[docs] def set_dirty(self, type_cls, dirty): """Marks/unmarks a certain type as dirty. :param type_cls: A :class:`Node` subclass. :param dirty: Wether the type is dirty. :type dirty: bool """ if dirty and type_cls not in self._dirty: self._dirty.add(type_cls) elif not dirty and type_cls in self._dirty: self._dirty.remove(type_cls)
[docs] def refresh_types(self, type_clss): """Instructs the backend to refresh certain types. "type_clss" is a list of :class:`dbschema.objects.Node` classes that should be refreshed. """ pass
def _refresh_types_internal(self, type_clss): tbd = set(type_clss).intersection(self._dirty) if not tbd: return self._dirty = self._dirty.difference(tbd) self.refresh_types(tbd)
[docs]class Namespace(Node): """A namespace/schema in the database."""
[docs] def is_default_namespace(self): """Returns True if this namespace is a default namespace.""" return self.db.get_default_namespace() == self
[docs]class Table(Node): """A table in the database."""
[docs] def get_columns(self): """Yields columns of this table. :rtype: Generator of :class:`Column` instances. """ self.db._refresh_types_internal([Column]) return self.find(type_cls=Column)
[docs] def get_foreign_keys(self): """Yields foreign key definitions. :rtype: Generator of :class:`ForeignKey` instances. """ self.db._refresh_types_internal([ForeignKey]) return self.find(type_cls=ForeignKey)
[docs] def get_reverse_foreign_keys(self): """Yields foreign keys pointing to this table. :rtype: Generaotr of :class:`ForeignKey` instances. """ self.db._refresh_types_internal([ForeignKey]) return self.db.find(type_cls=ForeignKey, foreign_table=self)
[docs]class View(Node): """A view in the database."""
[docs] def get_columns(self): """Yields columns of this view.""" self.db._refresh_types_internal([Column]) return self.find(type_cls=Column)
[docs]class Column(Node): """A column of a view or table."""
[docs]class ForeignKey(Node): """A foreign key definition of a table.""" def __init__(self, *args, **kwargs): self.foreign_table = None super(ForeignKey, self).__init__(*args, **kwargs) def get_foreign_table(self): return self.foreign_table