schema.py

SQLAlchemy. 经典的Python ORM框架。学习必看。

    def append_column(self, col):
        self.columns.add(col)

    def copy(self):
        return UniqueConstraint(name=self.name, *self.__colnames)

class Index(SchemaItem):
    """Represent an index of columns from a database table."""

    def __init__(self, name, *columns, **kwargs):
        """Construct an index object.

        Arguments are:

        name
          The name of the index

        \*columns
          Columns to include in the index. All columns must belong to
          the same table, and no column may appear more than once.

        \**kwargs
          Keyword arguments include:

          unique
            Defaults to False: create a unique index.

          postgres_where
            Defaults to None: create a partial index when using PostgreSQL
        """

        self.name = name
        self.columns = []
        self.table = None
        self.unique = kwargs.pop('unique', False)
        self.kwargs = kwargs

        self._init_items(*columns)

    def _init_items(self, *args):
        for column in args:
            self.append_column(column)

    def _set_parent(self, table):
        self.table = table
        self.metadata = table.metadata
        table.indexes.add(self)

    def append_column(self, column):
        # make sure all columns are from the same table
        # and no column is repeated
        if self.table is None:
            self._set_parent(column.table)
        elif column.table != self.table:
            # all columns muse be from same table
            raise exceptions.ArgumentError("All index columns must be from same table. "
                                "%s is from %s not %s" % (column,
                                                          column.table,
                                                          self.table))
        elif column.name in [ c.name for c in self.columns ]:
            raise exceptions.ArgumentError("A column may not appear twice in the "
                                "same index (%s already has column %s)"
                                % (self.name, column))
        self.columns.append(column)

    def create(self, bind=None):
        if bind is None:
            bind = _bind_or_error(self)
        bind.create(self)
        return self

    def drop(self, bind=None):
        if bind is None:
            bind = _bind_or_error(self)
        bind.drop(self)

    def __str__(self):
        return repr(self)

    def __repr__(self):
        return 'Index("%s", %s%s)' % (self.name,
                                      ', '.join([repr(c)
                                                 for c in self.columns]),
                                      (self.unique and ', unique=True') or '')

class MetaData(SchemaItem):
    """A collection of Tables and their associated schema constructs.

    Holds a collection of Tables and an optional binding to an
    ``Engine`` or ``Connection``.  If bound, the
    [sqlalchemy.schema#Table] objects in the collection and their
    columns may participate in implicit SQL execution.

    The ``bind`` property may be assigned to dynamically.  A common
    pattern is to start unbound and then bind later when an engine is
    available::

      metadata = MetaData()
      # define tables
      Table('mytable', metadata, ...)
      # connect to an engine later, perhaps after loading a URL from a
      # configuration file
      metadata.bind = an_engine

    MetaData is a thread-safe object after tables have been explicitly
    defined or loaded via reflection.
    """

    __visit_name__ = 'metadata'

    ddl_events = ('before-create', 'after-create', 'before-drop', 'after-drop')

    def __init__(self, bind=None, reflect=False):
        """Create a new MetaData object.

        bind
          An Engine or Connection to bind to.  May also be a string or
          URL instance, these are passed to create_engine() and this
          MetaData will be bound to the resulting engine.

        reflect
          Optional, automatically load all tables from the bound database.
          Defaults to False. ``bind`` is required when this option is
          set.  For finer control over loaded tables, use the ``reflect``
          method of ``MetaData``.

        """

        self.tables = {}
        self.bind = bind
        self.metadata = self
        self.ddl_listeners = util.defaultdict(list)
        if reflect:
            if not bind:
                raise exceptions.ArgumentError(
                    "A bind must be supplied in conjunction with reflect=True")
            self.reflect()

    def __repr__(self):
        return 'MetaData(%r)' % self.bind

    def __contains__(self, key):
        return key in self.tables

    def __getstate__(self):
        return {'tables': self.tables}

    def __setstate__(self, state):
        self.tables = state['tables']
        self._bind = None

    def is_bound(self):
        """True if this MetaData is bound to an Engine or Connection."""

        return self._bind is not None

    # @deprecated
    def connect(self, bind, **kwargs):
        """Bind this MetaData to an Engine.

        Use ``metadata.bind = <engine>`` or ``metadata.bind = <url>``.

        bind
          A string, ``URL``, ``Engine`` or ``Connection`` instance.  If
          a string or ``URL``, will be passed to ``create_engine()`` along
          with ``\**kwargs`` to produce the engine which to connect to.
          Otherwise connects directly to the given ``Engine``.
        """

        global URL
        if URL is None:
            from sqlalchemy.engine.url import URL
        if isinstance(bind, (basestring, URL)):
            self._bind = sqlalchemy.create_engine(bind, **kwargs)
        else:
            self._bind = bind
    connect = util.deprecated(connect)

    def bind(self):
        """An Engine or Connection to which this MetaData is bound.

        This property may be assigned an ``Engine`` or
        ``Connection``, or assigned a string or URL to
        automatically create a basic ``Engine`` for this bind
        with ``create_engine()``.
        """

        return self._bind

    def _bind_to(self, bind):
        """Bind this MetaData to an Engine, Connection, string or URL."""

        global URL
        if URL is None:
            from sqlalchemy.engine.url import URL

        if isinstance(bind, (basestring, URL)):
            from sqlalchemy import create_engine
            self._bind = create_engine(bind)
        else:
            self._bind = bind
    bind = property(bind, _bind_to)

    def clear(self):
        self.tables.clear()

    def remove(self, table):
        # TODO: scan all other tables and remove FK _column
        del self.tables[table.key]

    def table_iterator(self, reverse=True, tables=None):
        from sqlalchemy.sql.util import sort_tables
        if tables is None:
            tables = self.tables.values()
        else:
            tables = util.Set(tables).intersection(self.tables.values())
        return iter(sort_tables(tables, reverse=reverse))

    def reflect(self, bind=None, schema=None, only=None):
        """Load all available table definitions from the database.

        Automatically creates ``Table`` entries in this ``MetaData`` for any
        table available in the database but not yet present in the ``MetaData``.
        May be called multiple times to pick up tables recently added to the
        database, however no special action is taken if a table in this
        ``MetaData`` no longer exists in the database.

        bind
          A ``Connectable`` used to access the database; if None, uses
          the existing bind on this ``MetaData``, if any.

        schema
          Optional, query and reflect tables from an alterate schema.

        only
          Optional.  Load only a sub-set of available named tables.  May
          be specified as a sequence of names or a callable.

          If a sequence of names is provided, only those tables will be
          reflected.  An error is raised if a table is requested but not
          available.  Named tables already present in this ``MetaData`` are
          ignored.

          If a callable is provided, it will be used as a boolean predicate
          to filter the list of potential table names.  The callable is
          called with a table name and this ``MetaData`` instance as positional
          arguments and should return a true value for any table to reflect.
        """

        reflect_opts = {'autoload': True}
        if bind is None:
            bind = _bind_or_error(self)
            conn = None
        else:
            reflect_opts['autoload_with'] = bind
            conn = bind.contextual_connect()

        if schema is not None:
            reflect_opts['schema'] = schema

        available = util.OrderedSet(bind.engine.table_names(schema,
                                                            connection=conn))
        current = util.Set(self.tables.keys())

        if only is None:
            load = [name for name in available if name not in current]
        elif callable(only):
            load = [name for name in available
                    if name not in current and only(name, self)]
        else:
            missing = [name for name in only if name not in available]
            if missing:
                s = schema and (" schema '%s'" % schema) or ''
                raise exceptions.InvalidRequestError(
                    'Could not reflect: requested table(s) not available '
                    'in %s%s: (%s)' % (bind.engine.url, s, ', '.join(missing)))
            load = [name for name in only if name not in current]

        for name in load:
            Table(name, self, **reflect_opts)

    def append_ddl_listener(self, event, listener):
        """Append a DDL event listener to this ``MetaData``.

        The ``listener`` callable will be triggered when this ``MetaData`` is
        involved in DDL creates or drops, and will be invoked either before
        all Table-related actions or after.

        Arguments are:

        event
          One of ``MetaData.ddl_events``; 'before-create', 'after-create',
          'before-drop' or 'after-drop'.
        listener
          A callable, invoked with three positional arguments:

          event
            The event currently being handled
          schema_item
            The ``MetaData`` object being operated upon
          bind
            The ``Connection`` bueing used for DDL execution.

        Listeners are added to the MetaData's ``ddl_listeners`` attribute.

        Note: MetaData listeners are invoked even when ``Tables`` are created
        in isolation.  This may change in a future release. I.e.::

          # triggers all MetaData and Table listeners:
          metadata.create_all()

          # triggers MetaData listeners too:
          some.table.create()
        """

        if event not in self.ddl_events:
            raise LookupError(event)
        self.ddl_listeners[event].append(listener)

    def create_all(self, bind=None, tables=None, checkfirst=True):
        """Create all tables stored in this metadata.

        This will conditionally create tables depending on if they do
        not yet exist in the database.

        bind
          A ``Connectable`` used to access the database; if None, uses
          the existing bind on this ``MetaData``, if any.

        tables
          Optional list of ``Table`` objects, which is a subset of the
          total tables in the ``MetaData`` (others are ignored).
        """

        if bind is None:
            bind = _bind_or_error(self)
        for listener in self.ddl_listeners['before-create']:
            listener('before-create', self, bind)
        bind.create(self, checkfirst=checkfirst, tables=tables)
        for listener in self.ddl_listeners['after-create']: