__init__.py

DataJoint for Python is a framework for building data pipelines using MySQL databases to represent pipeline structure and bulk storage systems for large objects. DataJoint is built on the foundation of the relational data model and prescribes a consistent method for organizing, populating, and querying data.

The DataJoint data model is described in https://arxiv.org/abs/1807.11104

DataJoint is free software under the LGPL License. In addition, we request that any use of DataJoint leading to a publication be acknowledged in the publication.

Please cite:

- http://biorxiv.org/content/early/2015/11/14/031658 - http://dx.doi.org/10.1101/031658

conn(host=None, user=None, password=None, *, init_fun=None, reset=False, use_tls=None)

Returns a persistent connection object to be shared by multiple modules. If the connection is not yet established or reset=True, a new connection is set up. If connection information is not provided, it is taken from config which takes the information from dj_local_conf.json. If the password is not specified in that file datajoint prompts for the password.

Parameters:

Name	Type	Description	Default
host		hostname	None
user		mysql user	None
password		mysql password	None
init_fun		initialization function	None
reset		whether the connection should be reset or not	False
use_tls		TLS encryption option. Valid options are: True (required), False (required no TLS), None (TLS preferred, default), dict (Manually specify values per https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#encrypted-connection-options).	None

Source code in datajoint/connection.py

def conn(
    host=None, user=None, password=None, *, init_fun=None, reset=False, use_tls=None
):
    """
    Returns a persistent connection object to be shared by multiple modules.
    If the connection is not yet established or reset=True, a new connection is set up.
    If connection information is not provided, it is taken from config which takes the
    information from dj_local_conf.json. If the password is not specified in that file
    datajoint prompts for the password.

    :param host: hostname
    :param user: mysql user
    :param password: mysql password
    :param init_fun: initialization function
    :param reset: whether the connection should be reset or not
    :param use_tls: TLS encryption option. Valid options are: True (required), False
        (required no TLS), None (TLS preferred, default), dict (Manually specify values per
        https://dev.mysql.com/doc/refman/5.7/en/connection-options.html#encrypted-connection-options).
    """
    if not hasattr(conn, "connection") or reset:
        host = host if host is not None else config["database.host"]
        user = user if user is not None else config["database.user"]
        password = password if password is not None else config["database.password"]
        if user is None:
            user = input("Please enter DataJoint username: ")
        if password is None:
            password = getpass(prompt="Please enter DataJoint password: ")
        init_fun = (
            init_fun if init_fun is not None else config["connection.init_function"]
        )
        use_tls = use_tls if use_tls is not None else config["database.use_tls"]
        conn.connection = Connection(host, user, password, None, init_fun, use_tls)
    return conn.connection

Connection

A dj.Connection object manages a connection to a database server. It also catalogues modules, schemas, tables, and their dependencies (foreign keys).

Most of the parameters below should be set in the local configuration file.

Parameters:

Name	Type	Description	Default
host		host name, may include port number as hostname:port, in which case it overrides the value in port	required
user		user name	required
password		password	required
port		port number	None
init_fun		connection initialization function (SQL)	None
use_tls		TLS encryption option	None

Source code in datajoint/connection.py

class Connection:
    """
    A dj.Connection object manages a connection to a database server.
    It also catalogues modules, schemas, tables, and their dependencies (foreign keys).

    Most of the parameters below should be set in the local configuration file.

    :param host: host name, may include port number as hostname:port, in which case it overrides the value in port
    :param user: user name
    :param password: password
    :param port: port number
    :param init_fun: connection initialization function (SQL)
    :param use_tls: TLS encryption option
    """

    def __init__(self, host, user, password, port=None, init_fun=None, use_tls=None):
        host_input, host = (host, get_host_hook(host))
        if ":" in host:
            # the port in the hostname overrides the port argument
            host, port = host.split(":")
            port = int(port)
        elif port is None:
            port = config["database.port"]
        self.conn_info = dict(host=host, port=port, user=user, passwd=password)
        if use_tls is not False:
            self.conn_info["ssl"] = (
                use_tls if isinstance(use_tls, dict) else {"ssl": {}}
            )
        self.conn_info["ssl_input"] = use_tls
        self.conn_info["host_input"] = host_input
        self.init_fun = init_fun
        logger.info("Connecting {user}@{host}:{port}".format(**self.conn_info))
        self._conn = None
        self._query_cache = None
        connect_host_hook(self)
        if self.is_connected:
            logger.info("Connected {user}@{host}:{port}".format(**self.conn_info))
            self.connection_id = self.query("SELECT connection_id()").fetchone()[0]
        else:
            raise errors.LostConnectionError("Connection failed.")
        self._in_transaction = False
        self.schemas = dict()
        self.dependencies = Dependencies(self)

    def __eq__(self, other):
        return self.conn_info == other.conn_info

    def __repr__(self):
        connected = "connected" if self.is_connected else "disconnected"
        return "DataJoint connection ({connected}) {user}@{host}:{port}".format(
            connected=connected, **self.conn_info
        )

    def connect(self):
        """Connect to the database server."""
        with warnings.catch_warnings():
            warnings.filterwarnings("ignore", ".*deprecated.*")
            try:
                self._conn = client.connect(
                    init_command=self.init_fun,
                    sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
                    "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION,ONLY_FULL_GROUP_BY",
                    charset=config["connection.charset"],
                    **{
                        k: v
                        for k, v in self.conn_info.items()
                        if k not in ["ssl_input", "host_input"]
                    },
                )
            except client.err.InternalError:
                self._conn = client.connect(
                    init_command=self.init_fun,
                    sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
                    "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION,ONLY_FULL_GROUP_BY",
                    charset=config["connection.charset"],
                    **{
                        k: v
                        for k, v in self.conn_info.items()
                        if not (
                            k in ["ssl_input", "host_input"]
                            or k == "ssl"
                            and self.conn_info["ssl_input"] is None
                        )
                    },
                )
        self._conn.autocommit(True)

    def set_query_cache(self, query_cache=None):
        """
        When query_cache is not None, the connection switches into the query caching mode, which entails:
        1. Only SELECT queries are allowed.
        2. The results of queries are cached under the path indicated by dj.config['query_cache']
        3. query_cache is a string that differentiates different cache states.

        :param query_cache: a string to initialize the hash for query results
        """
        self._query_cache = query_cache

    def purge_query_cache(self):
        """Purges all query cache."""
        if (
            isinstance(config.get(cache_key), str)
            and pathlib.Path(config[cache_key]).is_dir()
        ):
            for path in pathlib.Path(config[cache_key]).iterdir():
                if not path.is_dir():
                    path.unlink()

    def close(self):
        self._conn.close()

    def register(self, schema):
        self.schemas[schema.database] = schema
        self.dependencies.clear()

    def ping(self):
        """Ping the connection or raises an exception if the connection is closed."""
        self._conn.ping(reconnect=False)

    @property
    def is_connected(self):
        """Return true if the object is connected to the database server."""
        try:
            self.ping()
        except:
            return False
        return True

    @staticmethod
    def _execute_query(cursor, query, args, suppress_warnings):
        try:
            with warnings.catch_warnings():
                if suppress_warnings:
                    # suppress all warnings arising from underlying SQL library
                    warnings.simplefilter("ignore")
                cursor.execute(query, args)
        except client.err.Error as err:
            raise translate_query_error(err, query)

    def query(
        self, query, args=(), *, as_dict=False, suppress_warnings=True, reconnect=None
    ):
        """
        Execute the specified query and return the tuple generator (cursor).

        :param query: SQL query
        :param args: additional arguments for the client.cursor
        :param as_dict: If as_dict is set to True, the returned cursor objects returns
                        query results as dictionary.
        :param suppress_warnings: If True, suppress all warnings arising from underlying query library
        :param reconnect: when None, get from config, when True, attempt to reconnect if disconnected
        """
        # check cache first:
        use_query_cache = bool(self._query_cache)
        if use_query_cache and not re.match(r"\s*(SELECT|SHOW)", query):
            raise errors.DataJointError(
                "Only SELECT queries are allowed when query caching is on."
            )
        if use_query_cache:
            if not config[cache_key]:
                raise errors.DataJointError(
                    f"Provide filepath dj.config['{cache_key}'] when using query caching."
                )
            hash_ = uuid_from_buffer(
                (str(self._query_cache) + re.sub(r"`\$\w+`", "", query)).encode()
                + pack(args)
            )
            cache_path = pathlib.Path(config[cache_key]) / str(hash_)
            try:
                buffer = cache_path.read_bytes()
            except FileNotFoundError:
                pass  # proceed to query the database
            else:
                return EmulatedCursor(unpack(buffer))

        if reconnect is None:
            reconnect = config["database.reconnect"]
        logger.debug("Executing SQL:" + query[:query_log_max_length])
        cursor_class = client.cursors.DictCursor if as_dict else client.cursors.Cursor
        cursor = self._conn.cursor(cursor=cursor_class)
        try:
            self._execute_query(cursor, query, args, suppress_warnings)
        except errors.LostConnectionError:
            if not reconnect:
                raise
            logger.warning("MySQL server has gone away. Reconnecting to the server.")
            connect_host_hook(self)
            if self._in_transaction:
                self.cancel_transaction()
                raise errors.LostConnectionError(
                    "Connection was lost during a transaction."
                )
            logger.debug("Re-executing")
            cursor = self._conn.cursor(cursor=cursor_class)
            self._execute_query(cursor, query, args, suppress_warnings)

        if use_query_cache:
            data = cursor.fetchall()
            cache_path.write_bytes(pack(data))
            return EmulatedCursor(data)

        return cursor

    def get_user(self):
        """
        :return: the user name and host name provided by the client to the server.
        """
        return self.query("SELECT user()").fetchone()[0]

    # ---------- transaction processing
    @property
    def in_transaction(self):
        """
        :return: True if there is an open transaction.
        """
        self._in_transaction = self._in_transaction and self.is_connected
        return self._in_transaction

    def start_transaction(self):
        """
        Starts a transaction error.
        """
        if self.in_transaction:
            raise errors.DataJointError("Nested connections are not supported.")
        self.query("START TRANSACTION WITH CONSISTENT SNAPSHOT")
        self._in_transaction = True
        logger.debug("Transaction started")

    def cancel_transaction(self):
        """
        Cancels the current transaction and rolls back all changes made during the transaction.
        """
        self.query("ROLLBACK")
        self._in_transaction = False
        logger.debug("Transaction cancelled. Rolling back ...")

    def commit_transaction(self):
        """
        Commit all changes made during the transaction and close it.

        """
        self.query("COMMIT")
        self._in_transaction = False
        logger.debug("Transaction committed and closed.")

    # -------- context manager for transactions
    @property
    @contextmanager
    def transaction(self):
        """
        Context manager for transactions. Opens an transaction and closes it after the with statement.
        If an error is caught during the transaction, the commits are automatically rolled back.
        All errors are raised again.

        Example:
        >>> import datajoint as dj
        >>> with dj.conn().transaction as conn:
        >>>     # transaction is open here
        """
        try:
            self.start_transaction()
            yield self
        except:
            self.cancel_transaction()
            raise
        else:
            self.commit_transaction()

connect()

Connect to the database server.

Source code in datajoint/connection.py

def connect(self):
    """Connect to the database server."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", ".*deprecated.*")
        try:
            self._conn = client.connect(
                init_command=self.init_fun,
                sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
                "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION,ONLY_FULL_GROUP_BY",
                charset=config["connection.charset"],
                **{
                    k: v
                    for k, v in self.conn_info.items()
                    if k not in ["ssl_input", "host_input"]
                },
            )
        except client.err.InternalError:
            self._conn = client.connect(
                init_command=self.init_fun,
                sql_mode="NO_ZERO_DATE,NO_ZERO_IN_DATE,ERROR_FOR_DIVISION_BY_ZERO,"
                "STRICT_ALL_TABLES,NO_ENGINE_SUBSTITUTION,ONLY_FULL_GROUP_BY",
                charset=config["connection.charset"],
                **{
                    k: v
                    for k, v in self.conn_info.items()
                    if not (
                        k in ["ssl_input", "host_input"]
                        or k == "ssl"
                        and self.conn_info["ssl_input"] is None
                    )
                },
            )
    self._conn.autocommit(True)

set_query_cache(query_cache=None)

When query_cache is not None, the connection switches into the query caching mode, which entails: 1. Only SELECT queries are allowed. 2. The results of queries are cached under the path indicated by dj.config['query_cache'] 3. query_cache is a string that differentiates different cache states.

Parameters:

Name	Type	Description	Default
query_cache		a string to initialize the hash for query results	None

Source code in datajoint/connection.py

def set_query_cache(self, query_cache=None):
    """
    When query_cache is not None, the connection switches into the query caching mode, which entails:
    1. Only SELECT queries are allowed.
    2. The results of queries are cached under the path indicated by dj.config['query_cache']
    3. query_cache is a string that differentiates different cache states.

    :param query_cache: a string to initialize the hash for query results
    """
    self._query_cache = query_cache

purge_query_cache()

Purges all query cache.

Source code in datajoint/connection.py

def purge_query_cache(self):
    """Purges all query cache."""
    if (
        isinstance(config.get(cache_key), str)
        and pathlib.Path(config[cache_key]).is_dir()
    ):
        for path in pathlib.Path(config[cache_key]).iterdir():
            if not path.is_dir():
                path.unlink()

ping()

Ping the connection or raises an exception if the connection is closed.

Source code in datajoint/connection.py

def ping(self):
    """Ping the connection or raises an exception if the connection is closed."""
    self._conn.ping(reconnect=False)

is_connected property

Return true if the object is connected to the database server.

query(query, args=(), *, as_dict=False, suppress_warnings=True, reconnect=None)

Execute the specified query and return the tuple generator (cursor).

Parameters:

Name	Type	Description	Default
query		SQL query	required
args		additional arguments for the client.cursor	()
as_dict		If as_dict is set to True, the returned cursor objects returns query results as dictionary.	False
suppress_warnings		If True, suppress all warnings arising from underlying query library	True
reconnect		when None, get from config, when True, attempt to reconnect if disconnected	None

Source code in datajoint/connection.py

def query(
    self, query, args=(), *, as_dict=False, suppress_warnings=True, reconnect=None
):
    """
    Execute the specified query and return the tuple generator (cursor).

    :param query: SQL query
    :param args: additional arguments for the client.cursor
    :param as_dict: If as_dict is set to True, the returned cursor objects returns
                    query results as dictionary.
    :param suppress_warnings: If True, suppress all warnings arising from underlying query library
    :param reconnect: when None, get from config, when True, attempt to reconnect if disconnected
    """
    # check cache first:
    use_query_cache = bool(self._query_cache)
    if use_query_cache and not re.match(r"\s*(SELECT|SHOW)", query):
        raise errors.DataJointError(
            "Only SELECT queries are allowed when query caching is on."
        )
    if use_query_cache:
        if not config[cache_key]:
            raise errors.DataJointError(
                f"Provide filepath dj.config['{cache_key}'] when using query caching."
            )
        hash_ = uuid_from_buffer(
            (str(self._query_cache) + re.sub(r"`\$\w+`", "", query)).encode()
            + pack(args)
        )
        cache_path = pathlib.Path(config[cache_key]) / str(hash_)
        try:
            buffer = cache_path.read_bytes()
        except FileNotFoundError:
            pass  # proceed to query the database
        else:
            return EmulatedCursor(unpack(buffer))

    if reconnect is None:
        reconnect = config["database.reconnect"]
    logger.debug("Executing SQL:" + query[:query_log_max_length])
    cursor_class = client.cursors.DictCursor if as_dict else client.cursors.Cursor
    cursor = self._conn.cursor(cursor=cursor_class)
    try:
        self._execute_query(cursor, query, args, suppress_warnings)
    except errors.LostConnectionError:
        if not reconnect:
            raise
        logger.warning("MySQL server has gone away. Reconnecting to the server.")
        connect_host_hook(self)
        if self._in_transaction:
            self.cancel_transaction()
            raise errors.LostConnectionError(
                "Connection was lost during a transaction."
            )
        logger.debug("Re-executing")
        cursor = self._conn.cursor(cursor=cursor_class)
        self._execute_query(cursor, query, args, suppress_warnings)

    if use_query_cache:
        data = cursor.fetchall()
        cache_path.write_bytes(pack(data))
        return EmulatedCursor(data)

    return cursor

get_user()

Returns:

Type	Description
	the user name and host name provided by the client to the server.

Source code in datajoint/connection.py

def get_user(self):
    """
    :return: the user name and host name provided by the client to the server.
    """
    return self.query("SELECT user()").fetchone()[0]

in_transaction property

Returns:

Type	Description
	True if there is an open transaction.

start_transaction()

Starts a transaction error.

Source code in datajoint/connection.py

def start_transaction(self):
    """
    Starts a transaction error.
    """
    if self.in_transaction:
        raise errors.DataJointError("Nested connections are not supported.")
    self.query("START TRANSACTION WITH CONSISTENT SNAPSHOT")
    self._in_transaction = True
    logger.debug("Transaction started")

cancel_transaction()

Cancels the current transaction and rolls back all changes made during the transaction.

Source code in datajoint/connection.py

def cancel_transaction(self):
    """
    Cancels the current transaction and rolls back all changes made during the transaction.
    """
    self.query("ROLLBACK")
    self._in_transaction = False
    logger.debug("Transaction cancelled. Rolling back ...")

commit_transaction()

Commit all changes made during the transaction and close it.

Source code in datajoint/connection.py

def commit_transaction(self):
    """
    Commit all changes made during the transaction and close it.

    """
    self.query("COMMIT")
    self._in_transaction = False
    logger.debug("Transaction committed and closed.")

transaction property

Context manager for transactions. Opens an transaction and closes it after the with statement. If an error is caught during the transaction, the commits are automatically rolled back. All errors are raised again.

Example:

import datajoint as dj with dj.conn().transaction as conn: # transaction is open here

Schema

A schema object is a decorator for UserTable classes that binds them to their database. It also specifies the namespace context in which other UserTable classes are defined.

Source code in datajoint/schemas.py

class Schema:
    """
    A schema object is a decorator for UserTable classes that binds them to their database.
    It also specifies the namespace `context` in which other UserTable classes are defined.
    """

    def __init__(
        self,
        schema_name=None,
        context=None,
        *,
        connection=None,
        create_schema=True,
        create_tables=True,
        add_objects=None,
    ):
        """
        Associate database schema `schema_name`. If the schema does not exist, attempt to
        create it on the server.

        If the schema_name is omitted, then schema.activate(..) must be called later
        to associate with the database.

        :param schema_name: the database schema to associate.
        :param context: dictionary for looking up foreign key references, leave None to use local context.
        :param connection: Connection object. Defaults to datajoint.conn().
        :param create_schema: When False, do not create the schema and raise an error if missing.
        :param create_tables: When False, do not create tables and raise errors when accessing missing tables.
        :param add_objects: a mapping with additional objects to make available to the context in which table classes
        are declared.
        """
        self._log = None
        self.connection = connection
        self.database = None
        self.context = context
        self.create_schema = create_schema
        self.create_tables = create_tables
        self._jobs = None
        self.external = ExternalMapping(self)
        self.add_objects = add_objects
        self.declare_list = []
        if schema_name:
            self.activate(schema_name)

    def is_activated(self):
        return self.database is not None

    def activate(
        self,
        schema_name=None,
        *,
        connection=None,
        create_schema=None,
        create_tables=None,
        add_objects=None,
    ):
        """
        Associate database schema `schema_name`. If the schema does not exist, attempt to
        create it on the server.

        :param schema_name: the database schema to associate.
            schema_name=None is used to assert that the schema has already been activated.
        :param connection: Connection object. Defaults to datajoint.conn().
        :param create_schema: If False, do not create the schema and raise an error if missing.
        :param create_tables: If False, do not create tables and raise errors when attempting
            to access missing tables.
        :param add_objects: a mapping with additional objects to make available to the context
            in which table classes are declared.
        """
        if schema_name is None:
            if self.exists:
                return
            raise DataJointError("Please provide a schema_name to activate the schema.")
        if self.database is not None and self.exists:
            if self.database == schema_name:  # already activated
                return
            raise DataJointError(
                "The schema is already activated for schema {db}.".format(
                    db=self.database
                )
            )
        if connection is not None:
            self.connection = connection
        if self.connection is None:
            self.connection = conn()
        self.database = schema_name
        if create_schema is not None:
            self.create_schema = create_schema
        if create_tables is not None:
            self.create_tables = create_tables
        if add_objects:
            self.add_objects = add_objects
        if not self.exists:
            if not self.create_schema or not self.database:
                raise DataJointError(
                    "Database `{name}` has not yet been declared. "
                    "Set argument create_schema=True to create it.".format(
                        name=schema_name
                    )
                )
            # create database
            logger.debug("Creating schema `{name}`.".format(name=schema_name))
            try:
                self.connection.query(
                    "CREATE DATABASE `{name}`".format(name=schema_name)
                )
            except AccessError:
                raise DataJointError(
                    "Schema `{name}` does not exist and could not be created. "
                    "Check permissions.".format(name=schema_name)
                )
            else:
                self.log("created")
        self.connection.register(self)

        # decorate all tables already decorated
        for cls, context in self.declare_list:
            if self.add_objects:
                context = dict(context, **self.add_objects)
            self._decorate_master(cls, context)

    def _assert_exists(self, message=None):
        if not self.exists:
            raise DataJointError(
                message
                or "Schema `{db}` has not been created.".format(db=self.database)
            )

    def __call__(self, cls, *, context=None):
        """
        Binds the supplied class to a schema. This is intended to be used as a decorator.

        :param cls: class to decorate.
        :param context: supplied when called from spawn_missing_classes
        """
        context = context or self.context or inspect.currentframe().f_back.f_locals
        if issubclass(cls, Part):
            raise DataJointError(
                "The schema decorator should not be applied to Part tables."
            )
        if self.is_activated():
            self._decorate_master(cls, context)
        else:
            self.declare_list.append((cls, context))
        return cls

    def _decorate_master(self, cls, context):
        """

        :param cls: the master class to process
        :param context: the class' declaration context
        """
        self._decorate_table(
            cls, context=dict(context, self=cls, **{cls.__name__: cls})
        )
        # Process part tables
        for part in ordered_dir(cls):
            if part[0].isupper():
                part = getattr(cls, part)
                if inspect.isclass(part) and issubclass(part, Part):
                    part._master = cls
                    # allow addressing master by name or keyword 'master'
                    self._decorate_table(
                        part,
                        context=dict(
                            context, master=cls, self=part, **{cls.__name__: cls}
                        ),
                    )

    def _decorate_table(self, table_class, context, assert_declared=False):
        """
        assign schema properties to the table class and declare the table
        """
        table_class.database = self.database
        table_class._connection = self.connection
        table_class._heading = Heading(
            table_info=dict(
                conn=self.connection,
                database=self.database,
                table_name=table_class.table_name,
                context=context,
            )
        )
        table_class._support = [table_class.full_table_name]
        table_class.declaration_context = context

        # instantiate the class, declare the table if not already
        instance = table_class()
        is_declared = instance.is_declared
        if not is_declared and not assert_declared and self.create_tables:
            instance.declare(context)
            self.connection.dependencies.clear()
        is_declared = is_declared or instance.is_declared

        # add table definition to the doc string
        if isinstance(table_class.definition, str):
            table_class.__doc__ = (
                (table_class.__doc__ or "")
                + "\nTable definition:\n\n"
                + table_class.definition
            )

        # fill values in Lookup tables from their contents property
        if (
            isinstance(instance, Lookup)
            and hasattr(instance, "contents")
            and is_declared
        ):
            contents = list(instance.contents)
            if len(contents) > len(instance):
                if instance.heading.has_autoincrement:
                    warnings.warn(
                        (
                            "Contents has changed but cannot be inserted because "
                            "{table} has autoincrement."
                        ).format(table=instance.__class__.__name__)
                    )
                else:
                    instance.insert(contents, skip_duplicates=True)

    @property
    def log(self):
        self._assert_exists()
        if self._log is None:
            self._log = Log(self.connection, self.database)
        return self._log

    def __repr__(self):
        return "Schema `{name}`\n".format(name=self.database)

    @property
    def size_on_disk(self):
        """
        :return: size of the entire schema in bytes
        """
        self._assert_exists()
        return int(
            self.connection.query(
                """
            SELECT SUM(data_length + index_length)
            FROM information_schema.tables WHERE table_schema='{db}'
            """.format(
                    db=self.database
                )
            ).fetchone()[0]
        )

    def spawn_missing_classes(self, context=None):
        """
        Creates the appropriate python user table classes from tables in the schema and places them
        in the context.

        :param context: alternative context to place the missing classes into, e.g. locals()
        """
        self._assert_exists()
        if context is None:
            if self.context is not None:
                context = self.context
            else:
                # if context is missing, use the calling namespace
                frame = inspect.currentframe().f_back
                context = frame.f_locals
                del frame
        tables = [
            row[0]
            for row in self.connection.query("SHOW TABLES in `%s`" % self.database)
            if lookup_class_name(
                "`{db}`.`{tab}`".format(db=self.database, tab=row[0]), context, 0
            )
            is None
        ]
        master_classes = (Lookup, Manual, Imported, Computed)
        part_tables = []
        for table_name in tables:
            class_name = to_camel_case(table_name)
            if class_name not in context:
                try:
                    cls = next(
                        cls
                        for cls in master_classes
                        if re.fullmatch(cls.tier_regexp, table_name)
                    )
                except StopIteration:
                    if re.fullmatch(Part.tier_regexp, table_name):
                        part_tables.append(table_name)
                else:
                    # declare and decorate master table classes
                    context[class_name] = self(
                        type(class_name, (cls,), dict()), context=context
                    )

        # attach parts to masters
        for table_name in part_tables:
            groups = re.fullmatch(Part.tier_regexp, table_name).groupdict()
            class_name = to_camel_case(groups["part"])
            try:
                master_class = context[to_camel_case(groups["master"])]
            except KeyError:
                raise DataJointError(
                    "The table %s does not follow DataJoint naming conventions"
                    % table_name
                )
            part_class = type(class_name, (Part,), dict(definition=...))
            part_class._master = master_class
            self._decorate_table(part_class, context=context, assert_declared=True)
            setattr(master_class, class_name, part_class)

    def drop(self, force=False):
        """
        Drop the associated schema if it exists
        """
        if not self.exists:
            logger.info(
                "Schema named `{database}` does not exist. Doing nothing.".format(
                    database=self.database
                )
            )
        elif (
            not config["safemode"]
            or force
            or user_choice(
                "Proceed to delete entire schema `%s`?" % self.database, default="no"
            )
            == "yes"
        ):
            logger.debug("Dropping `{database}`.".format(database=self.database))
            try:
                self.connection.query(
                    "DROP DATABASE `{database}`".format(database=self.database)
                )
                logger.debug(
                    "Schema `{database}` was dropped successfully.".format(
                        database=self.database
                    )
                )
            except AccessError:
                raise AccessError(
                    "An attempt to drop schema `{database}` "
                    "has failed. Check permissions.".format(database=self.database)
                )

    @property
    def exists(self):
        """
        :return: true if the associated schema exists on the server
        """
        if self.database is None:
            raise DataJointError("Schema must be activated first.")
        return bool(
            self.connection.query(
                "SELECT schema_name "
                "FROM information_schema.schemata "
                "WHERE schema_name = '{database}'".format(database=self.database)
            ).rowcount
        )

    @property
    def jobs(self):
        """
        schema.jobs provides a view of the job reservation table for the schema

        :return: jobs table
        """
        self._assert_exists()
        if self._jobs is None:
            self._jobs = JobTable(self.connection, self.database)
        return self._jobs

    @property
    def code(self):
        self._assert_exists()
        return self.save()

    def save(self, python_filename=None):
        """
        Generate the code for a module that recreates the schema.
        This method is in preparation for a future release and is not officially supported.

        :return: a string containing the body of a complete Python module defining this schema.
        """
        self.connection.dependencies.load()
        self._assert_exists()
        module_count = itertools.count()
        # add virtual modules for referenced modules with names vmod0, vmod1, ...
        module_lookup = collections.defaultdict(
            lambda: "vmod" + str(next(module_count))
        )
        db = self.database

        def make_class_definition(table):
            tier = _get_tier(table).__name__
            class_name = table.split(".")[1].strip("`")
            indent = ""
            if tier == "Part":
                class_name = class_name.split("__")[-1]
                indent += "    "
            class_name = to_camel_case(class_name)

            def replace(s):
                d, tabs = s.group(1), s.group(2)
                return ("" if d == db else (module_lookup[d] + ".")) + ".".join(
                    to_camel_case(tab) for tab in tabs.lstrip("__").split("__")
                )

            return ("" if tier == "Part" else "\n@schema\n") + (
                "{indent}class {class_name}(dj.{tier}):\n"
                '{indent}    definition = """\n'
                '{indent}    {defi}"""'
            ).format(
                class_name=class_name,
                indent=indent,
                tier=tier,
                defi=re.sub(
                    r"`([^`]+)`.`([^`]+)`",
                    replace,
                    FreeTable(self.connection, table).describe(),
                ).replace("\n", "\n    " + indent),
            )

        tables = self.connection.dependencies.topo_sort()
        body = "\n\n".join(make_class_definition(table) for table in tables)
        python_code = "\n\n".join(
            (
                '"""This module was auto-generated by datajoint from an existing schema"""',
                "import datajoint as dj\n\nschema = dj.Schema('{db}')".format(db=db),
                "\n".join(
                    "{module} = dj.VirtualModule('{module}', '{schema_name}')".format(
                        module=v, schema_name=k
                    )
                    for k, v in module_lookup.items()
                ),
                body,
            )
        )
        if python_filename is None:
            return python_code
        with open(python_filename, "wt") as f:
            f.write(python_code)

    def list_tables(self):
        """
        Return a list of all tables in the schema except tables with ~ in first character such
        as ~logs and ~job

        :return: A list of table names from the database schema.
        """
        self.connection.dependencies.load()
        return [
            t
            for d, t in (
                full_t.replace("`", "").split(".")
                for full_t in self.connection.dependencies.topo_sort()
            )
            if d == self.database
        ]

activate(schema_name=None, *, connection=None, create_schema=None, create_tables=None, add_objects=None)

Associate database schema schema_name. If the schema does not exist, attempt to create it on the server.

Parameters:

Name	Type	Description	Default
schema_name		the database schema to associate. schema_name=None is used to assert that the schema has already been activated.	None
connection		Connection object. Defaults to datajoint.conn().	None
create_schema		If False, do not create the schema and raise an error if missing.	None
create_tables		If False, do not create tables and raise errors when attempting to access missing tables.	None
add_objects		a mapping with additional objects to make available to the context in which table classes are declared.	None

Source code in datajoint/schemas.py

def activate(
    self,
    schema_name=None,
    *,
    connection=None,
    create_schema=None,
    create_tables=None,
    add_objects=None,
):
    """
    Associate database schema `schema_name`. If the schema does not exist, attempt to
    create it on the server.

    :param schema_name: the database schema to associate.
        schema_name=None is used to assert that the schema has already been activated.
    :param connection: Connection object. Defaults to datajoint.conn().
    :param create_schema: If False, do not create the schema and raise an error if missing.
    :param create_tables: If False, do not create tables and raise errors when attempting
        to access missing tables.
    :param add_objects: a mapping with additional objects to make available to the context
        in which table classes are declared.
    """
    if schema_name is None:
        if self.exists:
            return
        raise DataJointError("Please provide a schema_name to activate the schema.")
    if self.database is not None and self.exists:
        if self.database == schema_name:  # already activated
            return
        raise DataJointError(
            "The schema is already activated for schema {db}.".format(
                db=self.database
            )
        )
    if connection is not None:
        self.connection = connection
    if self.connection is None:
        self.connection = conn()
    self.database = schema_name
    if create_schema is not None:
        self.create_schema = create_schema
    if create_tables is not None:
        self.create_tables = create_tables
    if add_objects:
        self.add_objects = add_objects
    if not self.exists:
        if not self.create_schema or not self.database:
            raise DataJointError(
                "Database `{name}` has not yet been declared. "
                "Set argument create_schema=True to create it.".format(
                    name=schema_name
                )
            )
        # create database
        logger.debug("Creating schema `{name}`.".format(name=schema_name))
        try:
            self.connection.query(
                "CREATE DATABASE `{name}`".format(name=schema_name)
            )
        except AccessError:
            raise DataJointError(
                "Schema `{name}` does not exist and could not be created. "
                "Check permissions.".format(name=schema_name)
            )
        else:
            self.log("created")
    self.connection.register(self)

    # decorate all tables already decorated
    for cls, context in self.declare_list:
        if self.add_objects:
            context = dict(context, **self.add_objects)
        self._decorate_master(cls, context)

size_on_disk property

Returns:

Type	Description
	size of the entire schema in bytes

spawn_missing_classes(context=None)

Creates the appropriate python user table classes from tables in the schema and places them in the context.

Parameters:

Name	Type	Description	Default
context		alternative context to place the missing classes into, e.g. locals()	None

Source code in datajoint/schemas.py

def spawn_missing_classes(self, context=None):
    """
    Creates the appropriate python user table classes from tables in the schema and places them
    in the context.

    :param context: alternative context to place the missing classes into, e.g. locals()
    """
    self._assert_exists()
    if context is None:
        if self.context is not None:
            context = self.context
        else:
            # if context is missing, use the calling namespace
            frame = inspect.currentframe().f_back
            context = frame.f_locals
            del frame
    tables = [
        row[0]
        for row in self.connection.query("SHOW TABLES in `%s`" % self.database)
        if lookup_class_name(
            "`{db}`.`{tab}`".format(db=self.database, tab=row[0]), context, 0
        )
        is None
    ]
    master_classes = (Lookup, Manual, Imported, Computed)
    part_tables = []
    for table_name in tables:
        class_name = to_camel_case(table_name)
        if class_name not in context:
            try:
                cls = next(
                    cls
                    for cls in master_classes
                    if re.fullmatch(cls.tier_regexp, table_name)
                )
            except StopIteration:
                if re.fullmatch(Part.tier_regexp, table_name):
                    part_tables.append(table_name)
            else:
                # declare and decorate master table classes
                context[class_name] = self(
                    type(class_name, (cls,), dict()), context=context
                )

    # attach parts to masters
    for table_name in part_tables:
        groups = re.fullmatch(Part.tier_regexp, table_name).groupdict()
        class_name = to_camel_case(groups["part"])
        try:
            master_class = context[to_camel_case(groups["master"])]
        except KeyError:
            raise DataJointError(
                "The table %s does not follow DataJoint naming conventions"
                % table_name
            )
        part_class = type(class_name, (Part,), dict(definition=...))
        part_class._master = master_class
        self._decorate_table(part_class, context=context, assert_declared=True)
        setattr(master_class, class_name, part_class)

drop(force=False)

Drop the associated schema if it exists

Source code in datajoint/schemas.py

def drop(self, force=False):
    """
    Drop the associated schema if it exists
    """
    if not self.exists:
        logger.info(
            "Schema named `{database}` does not exist. Doing nothing.".format(
                database=self.database
            )
        )
    elif (
        not config["safemode"]
        or force
        or user_choice(
            "Proceed to delete entire schema `%s`?" % self.database, default="no"
        )
        == "yes"
    ):
        logger.debug("Dropping `{database}`.".format(database=self.database))
        try:
            self.connection.query(
                "DROP DATABASE `{database}`".format(database=self.database)
            )
            logger.debug(
                "Schema `{database}` was dropped successfully.".format(
                    database=self.database
                )
            )
        except AccessError:
            raise AccessError(
                "An attempt to drop schema `{database}` "
                "has failed. Check permissions.".format(database=self.database)
            )

exists property

Returns:

Type	Description
	true if the associated schema exists on the server

jobs property

schema.jobs provides a view of the job reservation table for the schema

Returns:

Type	Description
	jobs table

save(python_filename=None)

Generate the code for a module that recreates the schema. This method is in preparation for a future release and is not officially supported.

Returns:

Type	Description
	a string containing the body of a complete Python module defining this schema.

Source code in datajoint/schemas.py

def save(self, python_filename=None):
    """
    Generate the code for a module that recreates the schema.
    This method is in preparation for a future release and is not officially supported.

    :return: a string containing the body of a complete Python module defining this schema.
    """
    self.connection.dependencies.load()
    self._assert_exists()
    module_count = itertools.count()
    # add virtual modules for referenced modules with names vmod0, vmod1, ...
    module_lookup = collections.defaultdict(
        lambda: "vmod" + str(next(module_count))
    )
    db = self.database

    def make_class_definition(table):
        tier = _get_tier(table).__name__
        class_name = table.split(".")[1].strip("`")
        indent = ""
        if tier == "Part":
            class_name = class_name.split("__")[-1]
            indent += "    "
        class_name = to_camel_case(class_name)

        def replace(s):
            d, tabs = s.group(1), s.group(2)
            return ("" if d == db else (module_lookup[d] + ".")) + ".".join(
                to_camel_case(tab) for tab in tabs.lstrip("__").split("__")
            )

        return ("" if tier == "Part" else "\n@schema\n") + (
            "{indent}class {class_name}(dj.{tier}):\n"
            '{indent}    definition = """\n'
            '{indent}    {defi}"""'
        ).format(
            class_name=class_name,
            indent=indent,
            tier=tier,
            defi=re.sub(
                r"`([^`]+)`.`([^`]+)`",
                replace,
                FreeTable(self.connection, table).describe(),
            ).replace("\n", "\n    " + indent),
        )

    tables = self.connection.dependencies.topo_sort()
    body = "\n\n".join(make_class_definition(table) for table in tables)
    python_code = "\n\n".join(
        (
            '"""This module was auto-generated by datajoint from an existing schema"""',
            "import datajoint as dj\n\nschema = dj.Schema('{db}')".format(db=db),
            "\n".join(
                "{module} = dj.VirtualModule('{module}', '{schema_name}')".format(
                    module=v, schema_name=k
                )
                for k, v in module_lookup.items()
            ),
            body,
        )
    )
    if python_filename is None:
        return python_code
    with open(python_filename, "wt") as f:
        f.write(python_code)

list_tables()

Return a list of all tables in the schema except tables with ~ in first character such as ~logs and ~job

Returns:

Type	Description
	A list of table names from the database schema.

Source code in datajoint/schemas.py

def list_tables(self):
    """
    Return a list of all tables in the schema except tables with ~ in first character such
    as ~logs and ~job

    :return: A list of table names from the database schema.
    """
    self.connection.dependencies.load()
    return [
        t
        for d, t in (
            full_t.replace("`", "").split(".")
            for full_t in self.connection.dependencies.topo_sort()
        )
        if d == self.database
    ]

VirtualModule

Bases: ModuleType

A virtual module imitates a Python module representing a DataJoint schema from table definitions in the database. It declares the schema objects and a class for each table.

Source code in datajoint/schemas.py

class VirtualModule(types.ModuleType):
    """
    A virtual module imitates a Python module representing a DataJoint schema from table definitions in the database.
    It declares the schema objects and a class for each table.
    """

    def __init__(
        self,
        module_name,
        schema_name,
        *,
        create_schema=False,
        create_tables=False,
        connection=None,
        add_objects=None,
    ):
        """
        Creates a python module with the given name from the name of a schema on the server and
        automatically adds classes to it corresponding to the tables in the schema.

        :param module_name: displayed module name
        :param schema_name: name of the database in mysql
        :param create_schema: if True, create the schema on the database server
        :param create_tables: if True, module.schema can be used as the decorator for declaring new
        :param connection: a dj.Connection object to pass into the schema
        :param add_objects: additional objects to add to the module
        :return: the python module containing classes from the schema object and the table classes
        """
        super(VirtualModule, self).__init__(name=module_name)
        _schema = Schema(
            schema_name,
            create_schema=create_schema,
            create_tables=create_tables,
            connection=connection,
        )
        if add_objects:
            self.__dict__.update(add_objects)
        self.__dict__["schema"] = _schema
        _schema.spawn_missing_classes(context=self.__dict__)

list_schemas(connection=None)

Parameters:

Name	Type	Description	Default
connection		a dj.Connection object	None

Returns:

Type	Description
	list of all accessible schemas on the server

Source code in datajoint/schemas.py

def list_schemas(connection=None):
    """
    :param connection: a dj.Connection object
    :return: list of all accessible schemas on the server
    """
    return [
        r[0]
        for r in (connection or conn()).query(
            "SELECT schema_name "
            "FROM information_schema.schemata "
            'WHERE schema_name <> "information_schema"'
        )
    ]

Table

Bases: QueryExpression

Table is an abstract class that represents a table in the schema. It implements insert and delete methods and inherits query functionality. To make it a concrete class, override the abstract properties specifying the connection, table name, database, and definition.

Source code in datajoint/table.py

class Table(QueryExpression):
    """
    Table is an abstract class that represents a table in the schema.
    It implements insert and delete methods and inherits query functionality.
    To make it a concrete class, override the abstract properties specifying the connection,
    table name, database, and definition.
    """

    _table_name = None  # must be defined in subclass
    _log_ = None  # placeholder for the Log table object

    # These properties must be set by the schema decorator (schemas.py) at class level
    # or by FreeTable at instance level
    database = None
    declaration_context = None

    @property
    def table_name(self):
        return self._table_name

    @property
    def class_name(self):
        return self.__class__.__name__

    @property
    def definition(self):
        raise NotImplementedError(
            "Subclasses of Table must implement the `definition` property"
        )

    def declare(self, context=None):
        """
        Declare the table in the schema based on self.definition.

        :param context: the context for foreign key resolution. If None, foreign keys are
            not allowed.
        """
        if self.connection.in_transaction:
            raise DataJointError(
                "Cannot declare new tables inside a transaction, "
                "e.g. from inside a populate/make call"
            )
        # Enforce strict CamelCase #1150
        if not is_camel_case(self.class_name):
            raise DataJointError(
                "Table class name `{name}` is invalid. Please use CamelCase. ".format(
                    name=self.class_name
                )
                + "Classes defining tables should be formatted in strict CamelCase."
            )
        sql, external_stores = declare(self.full_table_name, self.definition, context)
        sql = sql.format(database=self.database)
        try:
            # declare all external tables before declaring main table
            for store in external_stores:
                self.connection.schemas[self.database].external[store]
            self.connection.query(sql)
        except AccessError:
            # skip if no create privilege
            pass
        else:
            self._log("Declared " + self.full_table_name)

    def alter(self, prompt=True, context=None):
        """
        Alter the table definition from self.definition
        """
        if self.connection.in_transaction:
            raise DataJointError(
                "Cannot update table declaration inside a transaction, "
                "e.g. from inside a populate/make call"
            )
        if context is None:
            frame = inspect.currentframe().f_back
            context = dict(frame.f_globals, **frame.f_locals)
            del frame
        old_definition = self.describe(context=context)
        sql, external_stores = alter(self.definition, old_definition, context)
        if not sql:
            if prompt:
                logger.warn("Nothing to alter.")
        else:
            sql = "ALTER TABLE {tab}\n\t".format(
                tab=self.full_table_name
            ) + ",\n\t".join(sql)
            if not prompt or user_choice(sql + "\n\nExecute?") == "yes":
                try:
                    # declare all external tables before declaring main table
                    for store in external_stores:
                        self.connection.schemas[self.database].external[store]
                    self.connection.query(sql)
                except AccessError:
                    # skip if no create privilege
                    pass
                else:
                    # reset heading
                    self.__class__._heading = Heading(
                        table_info=self.heading.table_info
                    )
                    if prompt:
                        logger.info("Table altered")
                    self._log("Altered " + self.full_table_name)

    def from_clause(self):
        """
        :return: the FROM clause of SQL SELECT statements.
        """
        return self.full_table_name

    def get_select_fields(self, select_fields=None):
        """
        :return: the selected attributes from the SQL SELECT statement.
        """
        return (
            "*" if select_fields is None else self.heading.project(select_fields).as_sql
        )

    def parents(self, primary=None, as_objects=False, foreign_key_info=False):
        """

        :param primary: if None, then all parents are returned. If True, then only foreign keys composed of
            primary key attributes are considered.  If False, return foreign keys including at least one
            secondary attribute.
        :param as_objects: if False, return table names. If True, return table objects.
        :param foreign_key_info: if True, each element in result also includes foreign key info.
        :return: list of parents as table names or table objects
            with (optional) foreign key information.
        """
        get_edge = self.connection.dependencies.parents
        nodes = [
            next(iter(get_edge(name).items())) if name.isdigit() else (name, props)
            for name, props in get_edge(self.full_table_name, primary).items()
        ]
        if as_objects:
            nodes = [(FreeTable(self.connection, name), props) for name, props in nodes]
        if not foreign_key_info:
            nodes = [name for name, props in nodes]
        return nodes

    def children(self, primary=None, as_objects=False, foreign_key_info=False):
        """
        :param primary: if None, then all children are returned. If True, then only foreign keys composed of
            primary key attributes are considered.  If False, return foreign keys including at least one
            secondary attribute.
        :param as_objects: if False, return table names. If True, return table objects.
        :param foreign_key_info: if True, each element in result also includes foreign key info.
        :return: list of children as table names or table objects
            with (optional) foreign key information.
        """
        get_edge = self.connection.dependencies.children
        nodes = [
            next(iter(get_edge(name).items())) if name.isdigit() else (name, props)
            for name, props in get_edge(self.full_table_name, primary).items()
        ]
        if as_objects:
            nodes = [(FreeTable(self.connection, name), props) for name, props in nodes]
        if not foreign_key_info:
            nodes = [name for name, props in nodes]
        return nodes

    def descendants(self, as_objects=False):
        """
        :param as_objects: False - a list of table names; True - a list of table objects.
        :return: list of tables descendants in topological order.
        """
        return [
            FreeTable(self.connection, node) if as_objects else node
            for node in self.connection.dependencies.descendants(self.full_table_name)
            if not node.isdigit()
        ]

    def ancestors(self, as_objects=False):
        """
        :param as_objects: False - a list of table names; True - a list of table objects.
        :return: list of tables ancestors in topological order.
        """
        return [
            FreeTable(self.connection, node) if as_objects else node
            for node in self.connection.dependencies.ancestors(self.full_table_name)
            if not node.isdigit()
        ]

    def parts(self, as_objects=False):
        """
        return part tables either as entries in a dict with foreign key information or a list of objects

        :param as_objects: if False (default), the output is a dict describing the foreign keys. If True, return table objects.
        """
        self.connection.dependencies.load(force=False)
        nodes = [
            node
            for node in self.connection.dependencies.nodes
            if not node.isdigit() and node.startswith(self.full_table_name[:-1] + "__")
        ]
        return [FreeTable(self.connection, c) for c in nodes] if as_objects else nodes

    @property
    def is_declared(self):
        """
        :return: True is the table is declared in the schema.
        """
        return (
            self.connection.query(
                'SHOW TABLES in `{database}` LIKE "{table_name}"'.format(
                    database=self.database, table_name=self.table_name
                )
            ).rowcount
            > 0
        )

    @property
    def full_table_name(self):
        """
        :return: full table name in the schema
        """
        return r"`{0:s}`.`{1:s}`".format(self.database, self.table_name)

    @property
    def _log(self):
        if self._log_ is None:
            self._log_ = Log(
                self.connection,
                database=self.database,
                skip_logging=self.table_name.startswith("~"),
            )
        return self._log_

    @property
    def external(self):
        return self.connection.schemas[self.database].external

    def update1(self, row):
        """
        ``update1`` updates one existing entry in the table.
        Caution: In DataJoint the primary modes for data manipulation is to ``insert`` and
        ``delete`` entire records since referential integrity works on the level of records,
        not fields. Therefore, updates are reserved for corrective operations outside of main
        workflow. Use UPDATE methods sparingly with full awareness of potential violations of
        assumptions.

        :param row: a ``dict`` containing the primary key values and the attributes to update.
            Setting an attribute value to None will reset it to the default value (if any).

        The primary key attributes must always be provided.

        Examples:

        >>> table.update1({'id': 1, 'value': 3})  # update value in record with id=1
        >>> table.update1({'id': 1, 'value': None})  # reset value to default
        """
        # argument validations
        if not isinstance(row, collections.abc.Mapping):
            raise DataJointError("The argument of update1 must be dict-like.")
        if not set(row).issuperset(self.primary_key):
            raise DataJointError(
                "The argument of update1 must supply all primary key values."
            )
        try:
            raise DataJointError(
                "Attribute `%s` not found."
                % next(k for k in row if k not in self.heading.names)
            )
        except StopIteration:
            pass  # ok
        if len(self.restriction):
            raise DataJointError("Update cannot be applied to a restricted table.")
        key = {k: row[k] for k in self.primary_key}
        if len(self & key) != 1:
            raise DataJointError("Update can only be applied to one existing entry.")
        # UPDATE query
        row = [
            self.__make_placeholder(k, v)
            for k, v in row.items()
            if k not in self.primary_key
        ]
        query = "UPDATE {table} SET {assignments} WHERE {where}".format(
            table=self.full_table_name,
            assignments=",".join("`%s`=%s" % r[:2] for r in row),
            where=make_condition(self, key, set()),
        )
        self.connection.query(query, args=list(r[2] for r in row if r[2] is not None))

    def insert1(self, row, **kwargs):
        """
        Insert one data record into the table. For ``kwargs``, see ``insert()``.

        :param row: a numpy record, a dict-like object, or an ordered sequence to be inserted
            as one row.
        """
        self.insert((row,), **kwargs)

    def insert(
        self,
        rows,
        replace=False,
        skip_duplicates=False,
        ignore_extra_fields=False,
        allow_direct_insert=None,
    ):
        """
        Insert a collection of rows.

        :param rows: Either (a) an iterable where an element is a numpy record, a
            dict-like object, a pandas.DataFrame, a sequence, or a query expression with
            the same heading as self, or (b) a pathlib.Path object specifying a path
            relative to the current directory with a CSV file, the contents of which
            will be inserted.
        :param replace: If True, replaces the existing tuple.
        :param skip_duplicates: If True, silently skip duplicate inserts.
        :param ignore_extra_fields: If False, fields that are not in the heading raise error.
        :param allow_direct_insert: Only applies in auto-populated tables. If False (default),
            insert may only be called from inside the make callback.

        Example:

            >>> Table.insert([
            >>>     dict(subject_id=7, species="mouse", date_of_birth="2014-09-01"),
            >>>     dict(subject_id=8, species="mouse", date_of_birth="2014-09-02")])
        """
        if isinstance(rows, pandas.DataFrame):
            # drop 'extra' synthetic index for 1-field index case -
            # frames with more advanced indices should be prepared by user.
            rows = rows.reset_index(
                drop=len(rows.index.names) == 1 and not rows.index.names[0]
            ).to_records(index=False)

        if isinstance(rows, Path):
            with open(rows, newline="") as data_file:
                rows = list(csv.DictReader(data_file, delimiter=","))

        # prohibit direct inserts into auto-populated tables
        if not allow_direct_insert and not getattr(self, "_allow_insert", True):
            raise DataJointError(
                "Inserts into an auto-populated table can only be done inside "
                "its make method during a populate call."
                " To override, set keyword argument allow_direct_insert=True."
            )

        if inspect.isclass(rows) and issubclass(rows, QueryExpression):
            rows = rows()  # instantiate if a class
        if isinstance(rows, QueryExpression):
            # insert from select
            if not ignore_extra_fields:
                try:
                    raise DataJointError(
                        "Attribute %s not found. To ignore extra attributes in insert, "
                        "set ignore_extra_fields=True."
                        % next(
                            name for name in rows.heading if name not in self.heading
                        )
                    )
                except StopIteration:
                    pass
            fields = list(name for name in rows.heading if name in self.heading)
            query = "{command} INTO {table} ({fields}) {select}{duplicate}".format(
                command="REPLACE" if replace else "INSERT",
                fields="`" + "`,`".join(fields) + "`",
                table=self.full_table_name,
                select=rows.make_sql(fields),
                duplicate=(
                    " ON DUPLICATE KEY UPDATE `{pk}`={table}.`{pk}`".format(
                        table=self.full_table_name, pk=self.primary_key[0]
                    )
                    if skip_duplicates
                    else ""
                ),
            )
            self.connection.query(query)
            return

        # collects the field list from first row (passed by reference)
        field_list = []
        rows = list(
            self.__make_row_to_insert(row, field_list, ignore_extra_fields)
            for row in rows
        )
        if rows:
            try:
                query = "{command} INTO {destination}(`{fields}`) VALUES {placeholders}{duplicate}".format(
                    command="REPLACE" if replace else "INSERT",
                    destination=self.from_clause(),
                    fields="`,`".join(field_list),
                    placeholders=",".join(
                        "(" + ",".join(row["placeholders"]) + ")" for row in rows
                    ),
                    duplicate=(
                        " ON DUPLICATE KEY UPDATE `{pk}`=`{pk}`".format(
                            pk=self.primary_key[0]
                        )
                        if skip_duplicates
                        else ""
                    ),
                )
                self.connection.query(
                    query,
                    args=list(
                        itertools.chain.from_iterable(
                            (v for v in r["values"] if v is not None) for r in rows
                        )
                    ),
                )
            except UnknownAttributeError as err:
                raise err.suggest(
                    "To ignore extra fields in insert, set ignore_extra_fields=True"
                )
            except DuplicateError as err:
                raise err.suggest(
                    "To ignore duplicate entries in insert, set skip_duplicates=True"
                )

    def delete_quick(self, get_count=False):
        """
        Deletes the table without cascading and without user prompt.
        If this table has populated dependent tables, this will fail.
        """
        query = "DELETE FROM " + self.full_table_name + self.where_clause()
        self.connection.query(query)
        count = (
            self.connection.query("SELECT ROW_COUNT()").fetchone()[0]
            if get_count
            else None
        )
        self._log(query[:255])
        return count

    def delete(
        self,
        transaction: bool = True,
        safemode: Union[bool, None] = None,
        force_parts: bool = False,
        force_masters: bool = False,
    ) -> int:
        """
        Deletes the contents of the table and its dependent tables, recursively.

        Args:
            transaction: If `True`, use of the entire delete becomes an atomic transaction.
                This is the default and recommended behavior. Set to `False` if this delete is
                nested within another transaction.
            safemode: If `True`, prohibit nested transactions and prompt to confirm. Default
                is `dj.config['safemode']`.
            force_parts: Delete from parts even when not deleting from their masters.
            force_masters: If `True`, include part/master pairs in the cascade.
                Default is `False`.

        Returns:
            Number of deleted rows (excluding those from dependent tables).

        Raises:
            DataJointError: Delete exceeds maximum number of delete attempts.
            DataJointError: When deleting within an existing transaction.
            DataJointError: Deleting a part table before its master.
        """
        deleted = set()
        visited_masters = set()

        def cascade(table):
            """service function to perform cascading deletes recursively."""
            max_attempts = 50
            for _ in range(max_attempts):
                try:
                    delete_count = table.delete_quick(get_count=True)
                except IntegrityError as error:
                    match = foreign_key_error_regexp.match(error.args[0]).groupdict()
                    # if schema name missing, use table
                    if "`.`" not in match["child"]:
                        match["child"] = "{}.{}".format(
                            table.full_table_name.split(".")[0], match["child"]
                        )
                    if (
                        match["pk_attrs"] is not None
                    ):  # fully matched, adjusting the keys
                        match["fk_attrs"] = [
                            k.strip("`") for k in match["fk_attrs"].split(",")
                        ]
                        match["pk_attrs"] = [
                            k.strip("`") for k in match["pk_attrs"].split(",")
                        ]
                    else:  # only partially matched, querying with constraint to determine keys
                        match["fk_attrs"], match["parent"], match["pk_attrs"] = list(
                            map(
                                list,
                                zip(
                                    *table.connection.query(
                                        constraint_info_query,
                                        args=(
                                            match["name"].strip("`"),
                                            *[
                                                _.strip("`")
                                                for _ in match["child"].split("`.`")
                                            ],
                                        ),
                                    ).fetchall()
                                ),
                            )
                        )
                        match["parent"] = match["parent"][0]

                    # Restrict child by table if
                    #   1. if table's restriction attributes are not in child's primary key
                    #   2. if child renames any attributes
                    # Otherwise restrict child by table's restriction.
                    child = FreeTable(table.connection, match["child"])
                    if (
                        set(table.restriction_attributes) <= set(child.primary_key)
                        and match["fk_attrs"] == match["pk_attrs"]
                    ):
                        child._restriction = table._restriction
                        child._restriction_attributes = table.restriction_attributes
                    elif match["fk_attrs"] != match["pk_attrs"]:
                        child &= table.proj(
                            **dict(zip(match["fk_attrs"], match["pk_attrs"]))
                        )
                    else:
                        child &= table.proj()

                    master_name = get_master(child.full_table_name)
                    if (
                        force_masters
                        and master_name
                        and master_name != table.full_table_name
                        and master_name not in visited_masters
                    ):
                        master = FreeTable(table.connection, master_name)
                        master._restriction_attributes = set()
                        master._restriction = [
                            make_condition(  # &= may cause in target tables in subquery
                                master,
                                (master.proj() & child.proj()).fetch(),
                                master._restriction_attributes,
                            )
                        ]
                        visited_masters.add(master_name)
                        cascade(master)
                    else:
                        cascade(child)
                else:
                    deleted.add(table.full_table_name)
                    logger.info(
                        "Deleting {count} rows from {table}".format(
                            count=delete_count, table=table.full_table_name
                        )
                    )
                    break
            else:
                raise DataJointError("Exceeded maximum number of delete attempts.")
            return delete_count

        safemode = config["safemode"] if safemode is None else safemode

        # Start transaction
        if transaction:
            if not self.connection.in_transaction:
                self.connection.start_transaction()
            else:
                if not safemode:
                    transaction = False
                else:
                    raise DataJointError(
                        "Delete cannot use a transaction within an ongoing transaction. "
                        "Set transaction=False or safemode=False)."
                    )

        # Cascading delete
        try:
            delete_count = cascade(self)
        except:
            if transaction:
                self.connection.cancel_transaction()
            raise

        if not force_parts:
            # Avoid deleting from child before master (See issue #151)
            for part in deleted:
                master = get_master(part)
                if master and master not in deleted:
                    if transaction:
                        self.connection.cancel_transaction()
                    raise DataJointError(
                        "Attempt to delete part table {part} before deleting from "
                        "its master {master} first.".format(part=part, master=master)
                    )

        # Confirm and commit
        if delete_count == 0:
            if safemode:
                logger.warn("Nothing to delete.")
            if transaction:
                self.connection.cancel_transaction()
        elif not transaction:
            logger.info("Delete completed")
        else:
            if not safemode or user_choice("Commit deletes?", default="no") == "yes":
                if transaction:
                    self.connection.commit_transaction()
                if safemode:
                    logger.info("Deletes committed.")
            else:
                if transaction:
                    self.connection.cancel_transaction()
                if safemode:
                    logger.warn("Deletes cancelled")
        return delete_count

    def drop_quick(self):
        """
        Drops the table without cascading to dependent tables and without user prompt.
        """
        if self.is_declared:
            query = "DROP TABLE %s" % self.full_table_name
            self.connection.query(query)
            logger.info("Dropped table %s" % self.full_table_name)
            self._log(query[:255])
        else:
            logger.info(
                "Nothing to drop: table %s is not declared" % self.full_table_name
            )

    def drop(self):
        """
        Drop the table and all tables that reference it, recursively.
        User is prompted for confirmation if config['safemode'] is set to True.
        """
        if self.restriction:
            raise DataJointError(
                "A table with an applied restriction cannot be dropped."
                " Call drop() on the unrestricted Table."
            )
        self.connection.dependencies.load()
        do_drop = True
        tables = [
            table
            for table in self.connection.dependencies.descendants(self.full_table_name)
            if not table.isdigit()
        ]

        # avoid dropping part tables without their masters: See issue #374
        for part in tables:
            master = get_master(part)
            if master and master not in tables:
                raise DataJointError(
                    "Attempt to drop part table {part} before dropping "
                    "its master. Drop {master} first.".format(part=part, master=master)
                )

        if config["safemode"]:
            for table in tables:
                logger.info(
                    table + " (%d tuples)" % len(FreeTable(self.connection, table))
                )
            do_drop = user_choice("Proceed?", default="no") == "yes"
        if do_drop:
            for table in reversed(tables):
                FreeTable(self.connection, table).drop_quick()
            logger.info("Tables dropped. Restart kernel.")

    @property
    def size_on_disk(self):
        """
        :return: size of data and indices in bytes on the storage device
        """
        ret = self.connection.query(
            'SHOW TABLE STATUS FROM `{database}` WHERE NAME="{table}"'.format(
                database=self.database, table=self.table_name
            ),
            as_dict=True,
        ).fetchone()
        return ret["Data_length"] + ret["Index_length"]

    def show_definition(self):
        raise AttributeError(
            "show_definition is deprecated. Use the describe method instead."
        )

    def describe(self, context=None, printout=False):
        """
        :return:  the definition string for the query using DataJoint DDL.
        """
        if context is None:
            frame = inspect.currentframe().f_back
            context = dict(frame.f_globals, **frame.f_locals)
            del frame
        if self.full_table_name not in self.connection.dependencies:
            self.connection.dependencies.load()
        parents = self.parents(foreign_key_info=True)
        in_key = True
        definition = (
            "# " + self.heading.table_status["comment"] + "\n"
            if self.heading.table_status["comment"]
            else ""
        )
        attributes_thus_far = set()
        attributes_declared = set()
        indexes = self.heading.indexes.copy()
        for attr in self.heading.attributes.values():
            if in_key and not attr.in_key:
                definition += "---\n"
                in_key = False
            attributes_thus_far.add(attr.name)
            do_include = True
            for parent_name, fk_props in parents:
                if attr.name in fk_props["attr_map"]:
                    do_include = False
                    if attributes_thus_far.issuperset(fk_props["attr_map"]):
                        # foreign key properties
                        try:
                            index_props = indexes.pop(tuple(fk_props["attr_map"]))
                        except KeyError:
                            index_props = ""
                        else:
                            index_props = [k for k, v in index_props.items() if v]
                            index_props = (
                                " [{}]".format(", ".join(index_props))
                                if index_props
                                else ""
                            )

                        if not fk_props["aliased"]:
                            # simple foreign key
                            definition += "->{props} {class_name}\n".format(
                                props=index_props,
                                class_name=lookup_class_name(parent_name, context)
                                or parent_name,
                            )
                        else:
                            # projected foreign key
                            definition += (
                                "->{props} {class_name}.proj({proj_list})\n".format(
                                    props=index_props,
                                    class_name=lookup_class_name(parent_name, context)
                                    or parent_name,
                                    proj_list=",".join(
                                        '{}="{}"'.format(attr, ref)
                                        for attr, ref in fk_props["attr_map"].items()
                                        if ref != attr
                                    ),
                                )
                            )
                            attributes_declared.update(fk_props["attr_map"])
            if do_include:
                attributes_declared.add(attr.name)
                definition += "%-20s : %-28s %s\n" % (
                    (
                        attr.name
                        if attr.default is None
                        else "%s=%s" % (attr.name, attr.default)
                    ),
                    "%s%s"
                    % (attr.type, " auto_increment" if attr.autoincrement else ""),
                    "# " + attr.comment if attr.comment else "",
                )
        # add remaining indexes
        for k, v in indexes.items():
            definition += "{unique}INDEX ({attrs})\n".format(
                unique="UNIQUE " if v["unique"] else "", attrs=", ".join(k)
            )
        if printout:
            logger.info("\n" + definition)
        return definition

    # --- private helper functions ----
    def __make_placeholder(self, name, value, ignore_extra_fields=False):
        """
        For a given attribute `name` with `value`, return its processed value or value placeholder
        as a string to be included in the query and the value, if any, to be submitted for
        processing by mysql API.

        :param name:  name of attribute to be inserted
        :param value: value of attribute to be inserted
        """
        if ignore_extra_fields and name not in self.heading:
            return None
        attr = self.heading[name]
        if attr.adapter:
            value = attr.adapter.put(value)
        if value is None or (attr.numeric and (value == "" or np.isnan(float(value)))):
            # set default value
            placeholder, value = "DEFAULT", None
        else:  # not NULL
            placeholder = "%s"
            if attr.uuid:
                if not isinstance(value, uuid.UUID):
                    try:
                        value = uuid.UUID(value)
                    except (AttributeError, ValueError):
                        raise DataJointError(
                            "badly formed UUID value {v} for attribute `{n}`".format(
                                v=value, n=name
                            )
                        )
                value = value.bytes
            elif attr.is_blob:
                value = blob.pack(value)
                value = (
                    self.external[attr.store].put(value).bytes
                    if attr.is_external
                    else value
                )
            elif attr.is_attachment:
                attachment_path = Path(value)
                if attr.is_external:
                    # value is hash of contents
                    value = (
                        self.external[attr.store]
                        .upload_attachment(attachment_path)
                        .bytes
                    )
                else:
                    # value is filename + contents
                    value = (
                        str.encode(attachment_path.name)
                        + b"\0"
                        + attachment_path.read_bytes()
                    )
            elif attr.is_filepath:
                value = self.external[attr.store].upload_filepath(value).bytes
            elif attr.numeric:
                value = str(int(value) if isinstance(value, bool) else value)
            elif attr.json:
                value = json.dumps(value)
        return name, placeholder, value

    def __make_row_to_insert(self, row, field_list, ignore_extra_fields):
        """
        Helper function for insert and update

        :param row:  A tuple to insert
        :return: a dict with fields 'names', 'placeholders', 'values'
        """

        def check_fields(fields):
            """
            Validates that all items in `fields` are valid attributes in the heading

            :param fields: field names of a tuple
            """
            if not field_list:
                if not ignore_extra_fields:
                    for field in fields:
                        if field not in self.heading:
                            raise KeyError(
                                "`{0:s}` is not in the table heading".format(field)
                            )
            elif set(field_list) != set(fields).intersection(self.heading.names):
                raise DataJointError("Attempt to insert rows with different fields.")

        if isinstance(row, np.void):  # np.array
            check_fields(row.dtype.fields)
            attributes = [
                self.__make_placeholder(name, row[name], ignore_extra_fields)
                for name in self.heading
                if name in row.dtype.fields
            ]
        elif isinstance(row, collections.abc.Mapping):  # dict-based
            check_fields(row)
            attributes = [
                self.__make_placeholder(name, row[name], ignore_extra_fields)
                for name in self.heading
                if name in row
            ]
        else:  # positional
            try:
                if len(row) != len(self.heading):
                    raise DataJointError(
                        "Invalid insert argument. Incorrect number of attributes: "
                        "{given} given; {expected} expected".format(
                            given=len(row), expected=len(self.heading)
                        )
                    )
            except TypeError:
                raise DataJointError("Datatype %s cannot be inserted" % type(row))
            else:
                attributes = [
                    self.__make_placeholder(name, value, ignore_extra_fields)
                    for name, value in zip(self.heading, row)
                ]
        if ignore_extra_fields:
            attributes = [a for a in attributes if a is not None]

        assert len(attributes), "Empty tuple"
        row_to_insert = dict(zip(("names", "placeholders", "values"), zip(*attributes)))
        if not field_list:
            # first row sets the composition of the field list
            field_list.extend(row_to_insert["names"])
        else:
            #  reorder attributes in row_to_insert to match field_list
            order = list(row_to_insert["names"].index(field) for field in field_list)
            row_to_insert["names"] = list(row_to_insert["names"][i] for i in order)
            row_to_insert["placeholders"] = list(
                row_to_insert["placeholders"][i] for i in order
            )
            row_to_insert["values"] = list(row_to_insert["values"][i] for i in order)
        return row_to_insert

declare(context=None)

Declare the table in the schema based on self.definition.

Parameters:

Name	Type	Description	Default
context		the context for foreign key resolution. If None, foreign keys are not allowed.	None

Source code in datajoint/table.py

def declare(self, context=None):
    """
    Declare the table in the schema based on self.definition.

    :param context: the context for foreign key resolution. If None, foreign keys are
        not allowed.
    """
    if self.connection.in_transaction:
        raise DataJointError(
            "Cannot declare new tables inside a transaction, "
            "e.g. from inside a populate/make call"
        )
    # Enforce strict CamelCase #1150
    if not is_camel_case(self.class_name):
        raise DataJointError(
            "Table class name `{name}` is invalid. Please use CamelCase. ".format(
                name=self.class_name
            )
            + "Classes defining tables should be formatted in strict CamelCase."
        )
    sql, external_stores = declare(self.full_table_name, self.definition, context)
    sql = sql.format(database=self.database)
    try:
        # declare all external tables before declaring main table
        for store in external_stores:
            self.connection.schemas[self.database].external[store]
        self.connection.query(sql)
    except AccessError:
        # skip if no create privilege
        pass
    else:
        self._log("Declared " + self.full_table_name)

alter(prompt=True, context=None)

Alter the table definition from self.definition

Source code in datajoint/table.py

def alter(self, prompt=True, context=None):
    """
    Alter the table definition from self.definition
    """
    if self.connection.in_transaction:
        raise DataJointError(
            "Cannot update table declaration inside a transaction, "
            "e.g. from inside a populate/make call"
        )
    if context is None:
        frame = inspect.currentframe().f_back
        context = dict(frame.f_globals, **frame.f_locals)
        del frame
    old_definition = self.describe(context=context)
    sql, external_stores = alter(self.definition, old_definition, context)
    if not sql:
        if prompt:
            logger.warn("Nothing to alter.")
    else:
        sql = "ALTER TABLE {tab}\n\t".format(
            tab=self.full_table_name
        ) + ",\n\t".join(sql)
        if not prompt or user_choice(sql + "\n\nExecute?") == "yes":
            try:
                # declare all external tables before declaring main table
                for store in external_stores:
                    self.connection.schemas[self.database].external[store]
                self.connection.query(sql)
            except AccessError:
                # skip if no create privilege
                pass
            else:
                # reset heading
                self.__class__._heading = Heading(
                    table_info=self.heading.table_info
                )
                if prompt:
                    logger.info("Table altered")
                self._log("Altered " + self.full_table_name)

from_clause()

Returns:

Type	Description
	the FROM clause of SQL SELECT statements.

Source code in datajoint/table.py

def from_clause(self):
    """
    :return: the FROM clause of SQL SELECT statements.
    """
    return self.full_table_name

get_select_fields(select_fields=None)

Returns:

Type	Description
	the selected attributes from the SQL SELECT statement.

Source code in datajoint/table.py

def get_select_fields(self, select_fields=None):
    """
    :return: the selected attributes from the SQL SELECT statement.
    """
    return (
        "*" if select_fields is None else self.heading.project(select_fields).as_sql
    )

parents(primary=None, as_objects=False, foreign_key_info=False)

Parameters:

Name	Type	Description	Default
primary		if None, then all parents are returned. If True, then only foreign keys composed of primary key attributes are considered. If False, return foreign keys including at least one secondary attribute.	None
as_objects		if False, return table names. If True, return table objects.	False
foreign_key_info		if True, each element in result also includes foreign key info.	False

Returns:

Type	Description
	list of parents as table names or table objects with (optional) foreign key information.

Source code in datajoint/table.py

def parents(self, primary=None, as_objects=False, foreign_key_info=False):
    """

    :param primary: if None, then all parents are returned. If True, then only foreign keys composed of
        primary key attributes are considered.  If False, return foreign keys including at least one
        secondary attribute.
    :param as_objects: if False, return table names. If True, return table objects.
    :param foreign_key_info: if True, each element in result also includes foreign key info.
    :return: list of parents as table names or table objects
        with (optional) foreign key information.
    """
    get_edge = self.connection.dependencies.parents
    nodes = [
        next(iter(get_edge(name).items())) if name.isdigit() else (name, props)
        for name, props in get_edge(self.full_table_name, primary).items()
    ]
    if as_objects:
        nodes = [(FreeTable(self.connection, name), props) for name, props in nodes]
    if not foreign_key_info:
        nodes = [name for name, props in nodes]
    return nodes

children(primary=None, as_objects=False, foreign_key_info=False)

Parameters:

Name	Type	Description	Default
primary		if None, then all children are returned. If True, then only foreign keys composed of primary key attributes are considered. If False, return foreign keys including at least one secondary attribute.	None
as_objects		if False, return table names. If True, return table objects.	False
foreign_key_info		if True, each element in result also includes foreign key info.	False

Returns:

Type	Description
	list of children as table names or table objects with (optional) foreign key information.

Source code in datajoint/table.py

def children(self, primary=None, as_objects=False, foreign_key_info=False):
    """
    :param primary: if None, then all children are returned. If True, then only foreign keys composed of
        primary key attributes are considered.  If False, return foreign keys including at least one
        secondary attribute.
    :param as_objects: if False, return table names. If True, return table objects.
    :param foreign_key_info: if True, each element in result also includes foreign key info.
    :return: list of children as table names or table objects
        with (optional) foreign key information.
    """
    get_edge = self.connection.dependencies.children
    nodes = [
        next(iter(get_edge(name).items())) if name.isdigit() else (name, props)
        for name, props in get_edge(self.full_table_name, primary).items()
    ]
    if as_objects:
        nodes = [(FreeTable(self.connection, name), props) for name, props in nodes]
    if not foreign_key_info:
        nodes = [name for name, props in nodes]
    return nodes

descendants(as_objects=False)

Parameters:

Name	Type	Description	Default
as_objects		False - a list of table names; True - a list of table objects.	False

Returns:

Type	Description
	list of tables descendants in topological order.

Source code in datajoint/table.py

def descendants(self, as_objects=False):
    """
    :param as_objects: False - a list of table names; True - a list of table objects.
    :return: list of tables descendants in topological order.
    """
    return [
        FreeTable(self.connection, node) if as_objects else node
        for node in self.connection.dependencies.descendants(self.full_table_name)
        if not node.isdigit()
    ]

ancestors(as_objects=False)

Parameters:

Name	Type	Description	Default
as_objects		False - a list of table names; True - a list of table objects.	False

Returns:

Type	Description
	list of tables ancestors in topological order.

Source code in datajoint/table.py

def ancestors(self, as_objects=False):
    """
    :param as_objects: False - a list of table names; True - a list of table objects.
    :return: list of tables ancestors in topological order.
    """
    return [
        FreeTable(self.connection, node) if as_objects else node
        for node in self.connection.dependencies.ancestors(self.full_table_name)
        if not node.isdigit()
    ]

parts(as_objects=False)

return part tables either as entries in a dict with foreign key information or a list of objects

Parameters:

Name	Type	Description	Default
as_objects		if False (default), the output is a dict describing the foreign keys. If True, return table objects.	False

Source code in datajoint/table.py

def parts(self, as_objects=False):
    """
    return part tables either as entries in a dict with foreign key information or a list of objects

    :param as_objects: if False (default), the output is a dict describing the foreign keys. If True, return table objects.
    """
    self.connection.dependencies.load(force=False)
    nodes = [
        node
        for node in self.connection.dependencies.nodes
        if not node.isdigit() and node.startswith(self.full_table_name[:-1] + "__")
    ]
    return [FreeTable(self.connection, c) for c in nodes] if as_objects else nodes

is_declared property

Returns:

Type	Description
	True is the table is declared in the schema.

full_table_name property

Returns:

Type	Description
	full table name in the schema

update1(row)

update1 updates one existing entry in the table. Caution: In DataJoint the primary modes for data manipulation is to insert and delete entire records since referential integrity works on the level of records, not fields. Therefore, updates are reserved for corrective operations outside of main workflow. Use UPDATE methods sparingly with full awareness of potential violations of assumptions.

Parameters:

Name	Type	Description	Default
row		a dict containing the primary key values and the attributes to update. Setting an attribute value to None will reset it to the default value (if any). The primary key attributes must always be provided. Examples: >>> table.update1({'id': 1, 'value': 3}) # update value in record with id=1 >>> table.update1({'id': 1, 'value': None}) # reset value to default	required

Source code in datajoint/table.py

def update1(self, row):
    """
    ``update1`` updates one existing entry in the table.
    Caution: In DataJoint the primary modes for data manipulation is to ``insert`` and
    ``delete`` entire records since referential integrity works on the level of records,
    not fields. Therefore, updates are reserved for corrective operations outside of main
    workflow. Use UPDATE methods sparingly with full awareness of potential violations of
    assumptions.

    :param row: a ``dict`` containing the primary key values and the attributes to update.
        Setting an attribute value to None will reset it to the default value (if any).

    The primary key attributes must always be provided.

    Examples:

    >>> table.update1({'id': 1, 'value': 3})  # update value in record with id=1
    >>> table.update1({'id': 1, 'value': None})  # reset value to default
    """
    # argument validations
    if not isinstance(row, collections.abc.Mapping):
        raise DataJointError("The argument of update1 must be dict-like.")
    if not set(row).issuperset(self.primary_key):
        raise DataJointError(
            "The argument of update1 must supply all primary key values."
        )
    try:
        raise DataJointError(
            "Attribute `%s` not found."
            % next(k for k in row if k not in self.heading.names)
        )
    except StopIteration:
        pass  # ok
    if len(self.restriction):
        raise DataJointError("Update cannot be applied to a restricted table.")
    key = {k: row[k] for k in self.primary_key}
    if len(self & key) != 1:
        raise DataJointError("Update can only be applied to one existing entry.")
    # UPDATE query
    row = [
        self.__make_placeholder(k, v)
        for k, v in row.items()
        if k not in self.primary_key
    ]
    query = "UPDATE {table} SET {assignments} WHERE {where}".format(
        table=self.full_table_name,
        assignments=",".join("`%s`=%s" % r[:2] for r in row),
        where=make_condition(self, key, set()),
    )
    self.connection.query(query, args=list(r[2] for r in row if r[2] is not None))

insert1(row, **kwargs)

Insert one data record into the table. For kwargs, see insert().

Parameters:

Name	Type	Description	Default
row		a numpy record, a dict-like object, or an ordered sequence to be inserted as one row.	required

Source code in datajoint/table.py

def insert1(self, row, **kwargs):
    """
    Insert one data record into the table. For ``kwargs``, see ``insert()``.

    :param row: a numpy record, a dict-like object, or an ordered sequence to be inserted
        as one row.
    """
    self.insert((row,), **kwargs)

insert(rows, replace=False, skip_duplicates=False, ignore_extra_fields=False, allow_direct_insert=None)

Insert a collection of rows.

Parameters:

Name	Type	Description	Default
rows		Either (a) an iterable where an element is a numpy record, a dict-like object, a pandas.DataFrame, a sequence, or a query expression with the same heading as self, or (b) a pathlib.Path object specifying a path relative to the current directory with a CSV file, the contents of which will be inserted.	required
replace		If True, replaces the existing tuple.	False
skip_duplicates		If True, silently skip duplicate inserts.	False
ignore_extra_fields		If False, fields that are not in the heading raise error.	False
allow_direct_insert		Only applies in auto-populated tables. If False (default), insert may only be called from inside the make callback. Example: >>> Table.insert([ >>> dict(subject_id=7, species="mouse", date_of_birth="2014-09-01"), >>> dict(subject_id=8, species="mouse", date_of_birth="2014-09-02")])	None

Source code in datajoint/table.py

def insert(
    self,
    rows,
    replace=False,
    skip_duplicates=False,
    ignore_extra_fields=False,
    allow_direct_insert=None,
):
    """
    Insert a collection of rows.

    :param rows: Either (a) an iterable where an element is a numpy record, a
        dict-like object, a pandas.DataFrame, a sequence, or a query expression with
        the same heading as self, or (b) a pathlib.Path object specifying a path
        relative to the current directory with a CSV file, the contents of which
        will be inserted.
    :param replace: If True, replaces the existing tuple.
    :param skip_duplicates: If True, silently skip duplicate inserts.
    :param ignore_extra_fields: If False, fields that are not in the heading raise error.
    :param allow_direct_insert: Only applies in auto-populated tables. If False (default),
        insert may only be called from inside the make callback.

    Example:

        >>> Table.insert([
        >>>     dict(subject_id=7, species="mouse", date_of_birth="2014-09-01"),
        >>>     dict(subject_id=8, species="mouse", date_of_birth="2014-09-02")])
    """
    if isinstance(rows, pandas.DataFrame):
        # drop 'extra' synthetic index for 1-field index case -
        # frames with more advanced indices should be prepared by user.
        rows = rows.reset_index(
            drop=len(rows.index.names) == 1 and not rows.index.names[0]
        ).to_records(index=False)

    if isinstance(rows, Path):
        with open(rows, newline="") as data_file:
            rows = list(csv.DictReader(data_file, delimiter=","))

    # prohibit direct inserts into auto-populated tables
    if not allow_direct_insert and not getattr(self, "_allow_insert", True):
        raise DataJointError(
            "Inserts into an auto-populated table can only be done inside "
            "its make method during a populate call."
            " To override, set keyword argument allow_direct_insert=True."
        )

    if inspect.isclass(rows) and issubclass(rows, QueryExpression):
        rows = rows()  # instantiate if a class
    if isinstance(rows, QueryExpression):
        # insert from select
        if not ignore_extra_fields:
            try:
                raise DataJointError(
                    "Attribute %s not found. To ignore extra attributes in insert, "
                    "set ignore_extra_fields=True."
                    % next(
                        name for name in rows.heading if name not in self.heading
                    )
                )
            except StopIteration:
                pass
        fields = list(name for name in rows.heading if name in self.heading)
        query = "{command} INTO {table} ({fields}) {select}{duplicate}".format(
            command="REPLACE" if replace else "INSERT",
            fields="`" + "`,`".join(fields) + "`",
            table=self.full_table_name,
            select=rows.make_sql(fields),
            duplicate=(
                " ON DUPLICATE KEY UPDATE `{pk}`={table}.`{pk}`".format(
                    table=self.full_table_name, pk=self.primary_key[0]
                )
                if skip_duplicates
                else ""
            ),
        )
        self.connection.query(query)
        return

    # collects the field list from first row (passed by reference)
    field_list = []
    rows = list(
        self.__make_row_to_insert(row, field_list, ignore_extra_fields)
        for row in rows
    )
    if rows:
        try:
            query = "{command} INTO {destination}(`{fields}`) VALUES {placeholders}{duplicate}".format(
                command="REPLACE" if replace else "INSERT",
                destination=self.from_clause(),
                fields="`,`".join(field_list),
                placeholders=",".join(
                    "(" + ",".join(row["placeholders"]) + ")" for row in rows
                ),
                duplicate=(
                    " ON DUPLICATE KEY UPDATE `{pk}`=`{pk}`".format(
                        pk=self.primary_key[0]
                    )
                    if skip_duplicates
                    else ""
                ),
            )
            self.connection.query(
                query,
                args=list(
                    itertools.chain.from_iterable(
                        (v for v in r["values"] if v is not None) for r in rows
                    )
                ),
            )
        except UnknownAttributeError as err:
            raise err.suggest(
                "To ignore extra fields in insert, set ignore_extra_fields=True"
            )
        except DuplicateError as err:
            raise err.suggest(
                "To ignore duplicate entries in insert, set skip_duplicates=True"
            )

delete_quick(get_count=False)

Deletes the table without cascading and without user prompt. If this table has populated dependent tables, this will fail.

Source code in datajoint/table.py

def delete_quick(self, get_count=False):
    """
    Deletes the table without cascading and without user prompt.
    If this table has populated dependent tables, this will fail.
    """
    query = "DELETE FROM " + self.full_table_name + self.where_clause()
    self.connection.query(query)
    count = (
        self.connection.query("SELECT ROW_COUNT()").fetchone()[0]
        if get_count
        else None
    )
    self._log(query[:255])
    return count

delete(transaction=True, safemode=None, force_parts=False, force_masters=False)

Deletes the contents of the table and its dependent tables, recursively.

Args: transaction: If True, use of the entire delete becomes an atomic transaction. This is the default and recommended behavior. Set to False if this delete is nested within another transaction. safemode: If True, prohibit nested transactions and prompt to confirm. Default is dj.config['safemode']. force_parts: Delete from parts even when not deleting from their masters. force_masters: If True, include part/master pairs in the cascade. Default is False.

Returns: Number of deleted rows (excluding those from dependent tables).

Raises: DataJointError: Delete exceeds maximum number of delete attempts. DataJointError: When deleting within an existing transaction. DataJointError: Deleting a part table before its master.

Source code in datajoint/table.py

def delete(
    self,
    transaction: bool = True,
    safemode: Union[bool, None] = None,
    force_parts: bool = False,
    force_masters: bool = False,
) -> int:
    """
    Deletes the contents of the table and its dependent tables, recursively.

    Args:
        transaction: If `True`, use of the entire delete becomes an atomic transaction.
            This is the default and recommended behavior. Set to `False` if this delete is
            nested within another transaction.
        safemode: If `True`, prohibit nested transactions and prompt to confirm. Default
            is `dj.config['safemode']`.
        force_parts: Delete from parts even when not deleting from their masters.
        force_masters: If `True`, include part/master pairs in the cascade.
            Default is `False`.

    Returns:
        Number of deleted rows (excluding those from dependent tables).

    Raises:
        DataJointError: Delete exceeds maximum number of delete attempts.
        DataJointError: When deleting within an existing transaction.
        DataJointError: Deleting a part table before its master.
    """
    deleted = set()
    visited_masters = set()

    def cascade(table):
        """service function to perform cascading deletes recursively."""
        max_attempts = 50
        for _ in range(max_attempts):
            try:
                delete_count = table.delete_quick(get_count=True)
            except IntegrityError as error:
                match = foreign_key_error_regexp.match(error.args[0]).groupdict()
                # if schema name missing, use table
                if "`.`" not in match["child"]:
                    match["child"] = "{}.{}".format(
                        table.full_table_name.split(".")[0], match["child"]
                    )
                if (
                    match["pk_attrs"] is not None
                ):  # fully matched, adjusting the keys
                    match["fk_attrs"] = [
                        k.strip("`") for k in match["fk_attrs"].split(",")
                    ]
                    match["pk_attrs"] = [
                        k.strip("`") for k in match["pk_attrs"].split(",")
                    ]
                else:  # only partially matched, querying with constraint to determine keys
                    match["fk_attrs"], match["parent"], match["pk_attrs"] = list(
                        map(
                            list,
                            zip(
                                *table.connection.query(
                                    constraint_info_query,
                                    args=(
                                        match["name"].strip("`"),
                                        *[
                                            _.strip("`")
                                            for _ in match["child"].split("`.`")
                                        ],
                                    ),
                                ).fetchall()
                            ),
                        )
                    )
                    match["parent"] = match["parent"][0]

                # Restrict child by table if
                #   1. if table's restriction attributes are not in child's primary key
                #   2. if child renames any attributes
                # Otherwise restrict child by table's restriction.
                child = FreeTable(table.connection, match["child"])
                if (
                    set(table.restriction_attributes) <= set(child.primary_key)
                    and match["fk_attrs"] == match["pk_attrs"]
                ):
                    child._restriction = table._restriction
                    child._restriction_attributes = table.restriction_attributes
                elif match["fk_attrs"] != match["pk_attrs"]:
                    child &= table.proj(
                        **dict(zip(match["fk_attrs"], match["pk_attrs"]))
                    )
                else:
                    child &= table.proj()

                master_name = get_master(child.full_table_name)
                if (
                    force_masters
                    and master_name
                    and master_name != table.full_table_name
                    and master_name not in visited_masters
                ):
                    master = FreeTable(table.connection, master_name)
                    master._restriction_attributes = set()
                    master._restriction = [
                        make_condition(  # &= may cause in target tables in subquery
                            master,
                            (master.proj() & child.proj()).fetch(),
                            master._restriction_attributes,
                        )
                    ]
                    visited_masters.add(master_name)
                    cascade(master)
                else:
                    cascade(child)
            else:
                deleted.add(table.full_table_name)
                logger.info(
                    "Deleting {count} rows from {table}".format(
                        count=delete_count, table=table.full_table_name
                    )
                )
                break
        else:
            raise DataJointError("Exceeded maximum number of delete attempts.")
        return delete_count

    safemode = config["safemode"] if safemode is None else safemode

    # Start transaction
    if transaction:
        if not self.connection.in_transaction:
            self.connection.start_transaction()
        else:
            if not safemode:
                transaction = False
            else:
                raise DataJointError(
                    "Delete cannot use a transaction within an ongoing transaction. "
                    "Set transaction=False or safemode=False)."
                )

    # Cascading delete
    try:
        delete_count = cascade(self)
    except:
        if transaction:
            self.connection.cancel_transaction()
        raise

    if not force_parts:
        # Avoid deleting from child before master (See issue #151)
        for part in deleted:
            master = get_master(part)
            if master and master not in deleted:
                if transaction:
                    self.connection.cancel_transaction()
                raise DataJointError(
                    "Attempt to delete part table {part} before deleting from "
                    "its master {master} first.".format(part=part, master=master)
                )

    # Confirm and commit
    if delete_count == 0:
        if safemode:
            logger.warn("Nothing to delete.")
        if transaction:
            self.connection.cancel_transaction()
    elif not transaction:
        logger.info("Delete completed")
    else:
        if not safemode or user_choice("Commit deletes?", default="no") == "yes":
            if transaction:
                self.connection.commit_transaction()
            if safemode:
                logger.info("Deletes committed.")
        else:
            if transaction:
                self.connection.cancel_transaction()
            if safemode:
                logger.warn("Deletes cancelled")
    return delete_count

drop_quick()

Drops the table without cascading to dependent tables and without user prompt.

Source code in datajoint/table.py

def drop_quick(self):
    """
    Drops the table without cascading to dependent tables and without user prompt.
    """
    if self.is_declared:
        query = "DROP TABLE %s" % self.full_table_name
        self.connection.query(query)
        logger.info("Dropped table %s" % self.full_table_name)
        self._log(query[:255])
    else:
        logger.info(
            "Nothing to drop: table %s is not declared" % self.full_table_name
        )

drop()

Drop the table and all tables that reference it, recursively. User is prompted for confirmation if config['safemode'] is set to True.

Source code in datajoint/table.py

def drop(self):
    """
    Drop the table and all tables that reference it, recursively.
    User is prompted for confirmation if config['safemode'] is set to True.
    """
    if self.restriction:
        raise DataJointError(
            "A table with an applied restriction cannot be dropped."
            " Call drop() on the unrestricted Table."
        )
    self.connection.dependencies.load()
    do_drop = True
    tables = [
        table
        for table in self.connection.dependencies.descendants(self.full_table_name)
        if not table.isdigit()
    ]

    # avoid dropping part tables without their masters: See issue #374
    for part in tables:
        master = get_master(part)
        if master and master not in tables:
            raise DataJointError(
                "Attempt to drop part table {part} before dropping "
                "its master. Drop {master} first.".format(part=part, master=master)
            )

    if config["safemode"]:
        for table in tables:
            logger.info(
                table + " (%d tuples)" % len(FreeTable(self.connection, table))
            )
        do_drop = user_choice("Proceed?", default="no") == "yes"
    if do_drop:
        for table in reversed(tables):
            FreeTable(self.connection, table).drop_quick()
        logger.info("Tables dropped. Restart kernel.")

size_on_disk property

Returns:

Type	Description
	size of data and indices in bytes on the storage device

describe(context=None, printout=False)

Returns:

Type	Description
	the definition string for the query using DataJoint DDL.

Source code in datajoint/table.py

def describe(self, context=None, printout=False):
    """
    :return:  the definition string for the query using DataJoint DDL.
    """
    if context is None:
        frame = inspect.currentframe().f_back
        context = dict(frame.f_globals, **frame.f_locals)
        del frame
    if self.full_table_name not in self.connection.dependencies:
        self.connection.dependencies.load()
    parents = self.parents(foreign_key_info=True)
    in_key = True
    definition = (
        "# " + self.heading.table_status["comment"] + "\n"
        if self.heading.table_status["comment"]
        else ""
    )
    attributes_thus_far = set()
    attributes_declared = set()
    indexes = self.heading.indexes.copy()
    for attr in self.heading.attributes.values():
        if in_key and not attr.in_key:
            definition += "---\n"
            in_key = False
        attributes_thus_far.add(attr.name)
        do_include = True
        for parent_name, fk_props in parents:
            if attr.name in fk_props["attr_map"]:
                do_include = False
                if attributes_thus_far.issuperset(fk_props["attr_map"]):
                    # foreign key properties
                    try:
                        index_props = indexes.pop(tuple(fk_props["attr_map"]))
                    except KeyError:
                        index_props = ""
                    else:
                        index_props = [k for k, v in index_props.items() if v]
                        index_props = (
                            " [{}]".format(", ".join(index_props))
                            if index_props
                            else ""
                        )

                    if not fk_props["aliased"]:
                        # simple foreign key
                        definition += "->{props} {class_name}\n".format(
                            props=index_props,
                            class_name=lookup_class_name(parent_name, context)
                            or parent_name,
                        )
                    else:
                        # projected foreign key
                        definition += (
                            "->{props} {class_name}.proj({proj_list})\n".format(
                                props=index_props,
                                class_name=lookup_class_name(parent_name, context)
                                or parent_name,
                                proj_list=",".join(
                                    '{}="{}"'.format(attr, ref)
                                    for attr, ref in fk_props["attr_map"].items()
                                    if ref != attr
                                ),
                            )
                        )
                        attributes_declared.update(fk_props["attr_map"])
        if do_include:
            attributes_declared.add(attr.name)
            definition += "%-20s : %-28s %s\n" % (
                (
                    attr.name
                    if attr.default is None
                    else "%s=%s" % (attr.name, attr.default)
                ),
                "%s%s"
                % (attr.type, " auto_increment" if attr.autoincrement else ""),
                "# " + attr.comment if attr.comment else "",
            )
    # add remaining indexes
    for k, v in indexes.items():
        definition += "{unique}INDEX ({attrs})\n".format(
            unique="UNIQUE " if v["unique"] else "", attrs=", ".join(k)
        )
    if printout:
        logger.info("\n" + definition)
    return definition

FreeTable

Bases: Table

A base table without a dedicated class. Each instance is associated with a table specified by full_table_name.

Parameters:

Name	Type	Description	Default
conn		a dj.Connection object	required
full_table_name		in format database.table_name	required

Source code in datajoint/table.py

class FreeTable(Table):
    """
    A base table without a dedicated class. Each instance is associated with a table
    specified by full_table_name.

    :param conn:  a dj.Connection object
    :param full_table_name: in format `database`.`table_name`
    """

    def __init__(self, conn, full_table_name):
        self.database, self._table_name = (
            s.strip("`") for s in full_table_name.split(".")
        )
        self._connection = conn
        self._support = [full_table_name]
        self._heading = Heading(
            table_info=dict(
                conn=conn,
                database=self.database,
                table_name=self.table_name,
                context=None,
            )
        )

    def __repr__(self):
        return (
            "FreeTable(`%s`.`%s`)\n" % (self.database, self._table_name)
            + super().__repr__()
        )

Manual

Bases: UserTable

Inherit from this class if the table's values are entered manually.

Source code in datajoint/user_tables.py

class Manual(UserTable):
    """
    Inherit from this class if the table's values are entered manually.
    """

    _prefix = r""
    tier_regexp = r"(?P<manual>" + _prefix + _base_regexp + ")"

Lookup

Bases: UserTable

Inherit from this class if the table's values are for lookup. This is currently equivalent to defining the table as Manual and serves semantic purposes only.

Source code in datajoint/user_tables.py

class Lookup(UserTable):
    """
    Inherit from this class if the table's values are for lookup. This is
    currently equivalent to defining the table as Manual and serves semantic
    purposes only.
    """

    _prefix = "#"
    tier_regexp = (
        r"(?P<lookup>" + _prefix + _base_regexp.replace("TIER", "lookup") + ")"
    )

Imported

Bases: UserTable, AutoPopulate

Inherit from this class if the table's values are imported from external data sources. The inherited class must at least provide the function _make_tuples.

Source code in datajoint/user_tables.py

class Imported(UserTable, AutoPopulate):
    """
    Inherit from this class if the table's values are imported from external data sources.
    The inherited class must at least provide the function `_make_tuples`.
    """

    _prefix = "_"
    tier_regexp = r"(?P<imported>" + _prefix + _base_regexp + ")"

Computed

Bases: UserTable, AutoPopulate

Inherit from this class if the table's values are computed from other tables in the schema. The inherited class must at least provide the function _make_tuples.

Source code in datajoint/user_tables.py

class Computed(UserTable, AutoPopulate):
    """
    Inherit from this class if the table's values are computed from other tables in the schema.
    The inherited class must at least provide the function `_make_tuples`.
    """

    _prefix = "__"
    tier_regexp = r"(?P<computed>" + _prefix + _base_regexp + ")"

Part

Bases: UserTable

Inherit from this class if the table's values are details of an entry in another table and if this table is populated by the other table. For example, the entries inheriting from dj.Part could be single entries of a matrix, while the parent table refers to the entire matrix. Part tables are implemented as classes inside classes.

Source code in datajoint/user_tables.py

class Part(UserTable):
    """
    Inherit from this class if the table's values are details of an entry in another table
    and if this table is populated by the other table. For example, the entries inheriting from
    dj.Part could be single entries of a matrix, while the parent table refers to the entire matrix.
    Part tables are implemented as classes inside classes.
    """

    _connection = None
    _master = None

    tier_regexp = (
        r"(?P<master>"
        + "|".join([c.tier_regexp for c in (Manual, Lookup, Imported, Computed)])
        + r"){1,1}"
        + "__"
        + r"(?P<part>"
        + _base_regexp
        + ")"
    )

    @ClassProperty
    def connection(cls):
        return cls._connection

    @ClassProperty
    def full_table_name(cls):
        return (
            None
            if cls.database is None or cls.table_name is None
            else r"`{0:s}`.`{1:s}`".format(cls.database, cls.table_name)
        )

    @ClassProperty
    def master(cls):
        return cls._master

    @ClassProperty
    def table_name(cls):
        return (
            None
            if cls.master is None
            else cls.master.table_name + "__" + from_camel_case(cls.__name__)
        )

    def delete(self, force=False):
        """
        unless force is True, prohibits direct deletes from parts.
        """
        if force:
            super().delete(force_parts=True)
        else:
            raise DataJointError(
                "Cannot delete from a Part directly. Delete from master instead"
            )

    def drop(self, force=False):
        """
        unless force is True, prohibits direct deletes from parts.
        """
        if force:
            super().drop()
        else:
            raise DataJointError(
                "Cannot drop a Part directly.  Delete from master instead"
            )

    def alter(self, prompt=True, context=None):
        # without context, use declaration context which maps master keyword to master table
        super().alter(prompt=prompt, context=context or self.declaration_context)

delete(force=False)

unless force is True, prohibits direct deletes from parts.

Source code in datajoint/user_tables.py

def delete(self, force=False):
    """
    unless force is True, prohibits direct deletes from parts.
    """
    if force:
        super().delete(force_parts=True)
    else:
        raise DataJointError(
            "Cannot delete from a Part directly. Delete from master instead"
        )

drop(force=False)

unless force is True, prohibits direct deletes from parts.

Source code in datajoint/user_tables.py

def drop(self, force=False):
    """
    unless force is True, prohibits direct deletes from parts.
    """
    if force:
        super().drop()
    else:
        raise DataJointError(
            "Cannot drop a Part directly.  Delete from master instead"
        )

Not

invert restriction

Source code in datajoint/condition.py

class Not:
    """invert restriction"""

    def __init__(self, restriction):
        self.restriction = restriction

AndList

Bases: list

A list of conditions to by applied to a query expression by logical conjunction: the conditions are AND-ed. All other collections (lists, sets, other entity sets, etc) are applied by logical disjunction (OR).

Example: expr2 = expr & dj.AndList((cond1, cond2, cond3)) is equivalent to expr2 = expr & cond1 & cond2 & cond3

Source code in datajoint/condition.py

class AndList(list):
    """
    A list of conditions to by applied to a query expression by logical conjunction: the
    conditions are AND-ed. All other collections (lists, sets, other entity sets, etc) are
    applied by logical disjunction (OR).

    Example:
    expr2 = expr & dj.AndList((cond1, cond2, cond3))
    is equivalent to
    expr2 = expr & cond1 & cond2 & cond3
    """

    def append(self, restriction):
        if isinstance(restriction, AndList):
            # extend to reduce nesting
            self.extend(restriction)
        else:
            super().append(restriction)

U

dj.U objects are the universal sets representing all possible values of their attributes. dj.U objects cannot be queried on their own but are useful for forming some queries. dj.U('attr1', ..., 'attrn') represents the universal set with the primary key attributes attr1 ... attrn. The universal set is the set of all possible combinations of values of the attributes. Without any attributes, dj.U() represents the set with one element that has no attributes.

Restriction:

dj.U can be used to enumerate unique combinations of values of attributes from other expressions.

The following expression yields all unique combinations of contrast and brightness found in the stimulus set:

dj.U('contrast', 'brightness') & stimulus

Aggregation:

In aggregation, dj.U is used for summary calculation over an entire set:

The following expression yields one element with one attribute s containing the total number of elements in query expression expr:

dj.U().aggr(expr, n='count(*)')

The following expressions both yield one element containing the number n of distinct values of attribute attr in query expression expr.

dj.U().aggr(expr, n='count(distinct attr)') dj.U().aggr(dj.U('attr').aggr(expr), 'n=count(*)')

The following expression yields one element and one attribute s containing the sum of values of attribute attr over entire result set of expression expr:

dj.U().aggr(expr, s='sum(attr)')

The following expression yields the set of all unique combinations of attributes attr1, attr2 and the number of their occurrences in the result set of query expression expr.

dj.U(attr1,attr2).aggr(expr, n='count(*)')

Joins:

If expression expr has attributes 'attr1' and 'attr2', then expr * dj.U('attr1','attr2') yields the same result as expr but attr1 and attr2 are promoted to the the primary key. This is useful for producing a join on non-primary key attributes. For example, if attr is in both expr1 and expr2 but not in their primary keys, then expr1 * expr2 will throw an error because in most cases, it does not make sense to join on non-primary key attributes and users must first rename attr in one of the operands. The expression dj.U('attr') * rel1 * rel2 overrides this constraint.

Source code in datajoint/expression.py

class U:
    """
    dj.U objects are the universal sets representing all possible values of their attributes.
    dj.U objects cannot be queried on their own but are useful for forming some queries.
    dj.U('attr1', ..., 'attrn') represents the universal set with the primary key attributes attr1 ... attrn.
    The universal set is the set of all possible combinations of values of the attributes.
    Without any attributes, dj.U() represents the set with one element that has no attributes.

    Restriction:

    dj.U can be used to enumerate unique combinations of values of attributes from other expressions.

    The following expression yields all unique combinations of contrast and brightness found in the `stimulus` set:

    >>> dj.U('contrast', 'brightness') & stimulus

    Aggregation:

    In aggregation, dj.U is used for summary calculation over an entire set:

    The following expression yields one element with one attribute `s` containing the total number of elements in
    query expression `expr`:

    >>> dj.U().aggr(expr, n='count(*)')

    The following expressions both yield one element containing the number `n` of distinct values of attribute `attr` in
    query expression `expr`.

    >>> dj.U().aggr(expr, n='count(distinct attr)')
    >>> dj.U().aggr(dj.U('attr').aggr(expr), 'n=count(*)')

    The following expression yields one element and one attribute `s` containing the sum of values of attribute `attr`
    over entire result set of expression `expr`:

    >>> dj.U().aggr(expr, s='sum(attr)')

    The following expression yields the set of all unique combinations of attributes `attr1`, `attr2` and the number of
    their occurrences in the result set of query expression `expr`.

    >>> dj.U(attr1,attr2).aggr(expr, n='count(*)')

    Joins:

    If expression `expr` has attributes 'attr1' and 'attr2', then expr * dj.U('attr1','attr2') yields the same result
    as `expr` but `attr1` and `attr2` are promoted to the the primary key.  This is useful for producing a join on
    non-primary key attributes.
    For example, if `attr` is in both expr1 and expr2 but not in their primary keys, then expr1 * expr2 will throw
    an error because in most cases, it does not make sense to join on non-primary key attributes and users must first
    rename `attr` in one of the operands.  The expression dj.U('attr') * rel1 * rel2 overrides this constraint.
    """

    def __init__(self, *primary_key):
        self._primary_key = primary_key

    @property
    def primary_key(self):
        return self._primary_key

    def __and__(self, other):
        if inspect.isclass(other) and issubclass(other, QueryExpression):
            other = other()  # instantiate if a class
        if not isinstance(other, QueryExpression):
            raise DataJointError("Set U can only be restricted with a QueryExpression.")
        result = copy.copy(other)
        result._distinct = True
        result._heading = result.heading.set_primary_key(self.primary_key)
        result = result.proj()
        return result

    def join(self, other, left=False):
        """
        Joining U with a query expression has the effect of promoting the attributes of U to
        the primary key of the other query expression.

        :param other: the other query expression to join with.
        :param left: ignored. dj.U always acts as if left=False
        :return: a copy of the other query expression with the primary key extended.
        """
        if inspect.isclass(other) and issubclass(other, QueryExpression):
            other = other()  # instantiate if a class
        if not isinstance(other, QueryExpression):
            raise DataJointError("Set U can only be joined with a QueryExpression.")
        try:
            raise DataJointError(
                "Attribute `%s` not found"
                % next(k for k in self.primary_key if k not in other.heading.names)
            )
        except StopIteration:
            pass  # all ok
        result = copy.copy(other)
        result._heading = result.heading.set_primary_key(
            other.primary_key
            + [k for k in self.primary_key if k not in other.primary_key]
        )
        return result

    def __mul__(self, other):
        """shorthand for join"""
        return self.join(other)

    def aggr(self, group, **named_attributes):
        """
        Aggregation of the type U('attr1','attr2').aggr(group, computation="QueryExpression")
        has the primary key ('attr1','attr2') and performs aggregation computations for all matching elements of `group`.

        :param group:  The query expression to be aggregated.
        :param named_attributes: computations of the form new_attribute="sql expression on attributes of group"
        :return: The derived query expression
        """
        if named_attributes.get("keep_all_rows", False):
            raise DataJointError(
                "Cannot set keep_all_rows=True when aggregating on a universal set."
            )
        return Aggregation.create(self, group=group, keep_all_rows=False).proj(
            **named_attributes
        )

    aggregate = aggr  # alias for aggr

join(other, left=False)

Joining U with a query expression has the effect of promoting the attributes of U to the primary key of the other query expression.

Parameters:

Name	Type	Description	Default
other		the other query expression to join with.	required
left		ignored. dj.U always acts as if left=False	False

Returns:

Type	Description
	a copy of the other query expression with the primary key extended.

Source code in datajoint/expression.py

def join(self, other, left=False):
    """
    Joining U with a query expression has the effect of promoting the attributes of U to
    the primary key of the other query expression.

    :param other: the other query expression to join with.
    :param left: ignored. dj.U always acts as if left=False
    :return: a copy of the other query expression with the primary key extended.
    """
    if inspect.isclass(other) and issubclass(other, QueryExpression):
        other = other()  # instantiate if a class
    if not isinstance(other, QueryExpression):
        raise DataJointError("Set U can only be joined with a QueryExpression.")
    try:
        raise DataJointError(
            "Attribute `%s` not found"
            % next(k for k in self.primary_key if k not in other.heading.names)
        )
    except StopIteration:
        pass  # all ok
    result = copy.copy(other)
    result._heading = result.heading.set_primary_key(
        other.primary_key
        + [k for k in self.primary_key if k not in other.primary_key]
    )
    return result

aggr(group, **named_attributes)

Aggregation of the type U('attr1','attr2').aggr(group, computation="QueryExpression") has the primary key ('attr1','attr2') and performs aggregation computations for all matching elements of group.

Parameters:

Name	Type	Description	Default
group		The query expression to be aggregated.	required
named_attributes		computations of the form new_attribute="sql expression on attributes of group"	{}

Returns:

Type	Description
	The derived query expression

Source code in datajoint/expression.py

def aggr(self, group, **named_attributes):
    """
    Aggregation of the type U('attr1','attr2').aggr(group, computation="QueryExpression")
    has the primary key ('attr1','attr2') and performs aggregation computations for all matching elements of `group`.

    :param group:  The query expression to be aggregated.
    :param named_attributes: computations of the form new_attribute="sql expression on attributes of group"
    :return: The derived query expression
    """
    if named_attributes.get("keep_all_rows", False):
        raise DataJointError(
            "Cannot set keep_all_rows=True when aggregating on a universal set."
        )
    return Aggregation.create(self, group=group, keep_all_rows=False).proj(
        **named_attributes
    )

Top dataclass

A restriction to the top entities of a query. In SQL, this corresponds to ORDER BY ... LIMIT ... OFFSET

Source code in datajoint/condition.py

@dataclass
class Top:
    """
    A restriction to the top entities of a query.
    In SQL, this corresponds to ORDER BY ... LIMIT ... OFFSET
    """

    limit: Union[int, None] = 1
    order_by: Union[str, List[str]] = "KEY"
    offset: int = 0

    def __post_init__(self):
        self.order_by = self.order_by or ["KEY"]
        self.offset = self.offset or 0

        if self.limit is not None and not isinstance(self.limit, int):
            raise TypeError("Top limit must be an integer")
        if not isinstance(self.order_by, (str, collections.abc.Sequence)) or not all(
            isinstance(r, str) for r in self.order_by
        ):
            raise TypeError("Top order_by attributes must all be strings")
        if not isinstance(self.offset, int):
            raise TypeError("The offset argument must be an integer")
        if self.offset and self.limit is None:
            self.limit = 999999999999  # arbitrary large number to allow query
        if isinstance(self.order_by, str):
            self.order_by = [self.order_by]

Diagram

Bases: DiGraph

Schema diagram showing tables and foreign keys between in the form of a directed acyclic graph (DAG). The diagram is derived from the connection.dependencies object.

Usage:

diag = Diagram(source)

source can be a table object, a table class, a schema, or a module that has a schema.

diag.draw()

draws the diagram using pyplot

diag1 + diag2 - combines the two diagrams. diag1 - diag2 - difference between diagrams diag1 * diag2 - intersection of diagrams diag + n - expands n levels of successors diag - n - expands n levels of predecessors Thus dj.Diagram(schema.Table)+1-1 defines the diagram of immediate ancestors and descendants of schema.Table

Note that diagram + 1 - 1 may differ from diagram - 1 + 1 and so forth. Only those tables that are loaded in the connection object are displayed

Source code in datajoint/diagram.py

class Diagram(nx.DiGraph):
    """
    Schema diagram showing tables and foreign keys between in the form of a directed
    acyclic graph (DAG).  The diagram is derived from the connection.dependencies object.

    Usage:

    >>>  diag = Diagram(source)

    source can be a table object, a table class, a schema, or a module that has a schema.

    >>> diag.draw()

    draws the diagram using pyplot

    diag1 + diag2  - combines the two diagrams.
    diag1 - diag2  - difference between diagrams
    diag1 * diag2  - intersection of diagrams
    diag + n   - expands n levels of successors
    diag - n   - expands n levels of predecessors
    Thus dj.Diagram(schema.Table)+1-1 defines the diagram of immediate ancestors and descendants of schema.Table

    Note that diagram + 1 - 1  may differ from diagram - 1 + 1 and so forth.
    Only those tables that are loaded in the connection object are displayed
    """

    def __init__(self, source, context=None):

        if isinstance(source, Diagram):
            # copy constructor
            self.nodes_to_show = set(source.nodes_to_show)
            self.context = source.context
            super().__init__(source)
            return

        # get the caller's context
        if context is None:
            frame = inspect.currentframe().f_back
            self.context = dict(frame.f_globals, **frame.f_locals)
            del frame
        else:
            self.context = context

        # find connection in the source
        try:
            connection = source.connection
        except AttributeError:
            try:
                connection = source.schema.connection
            except AttributeError:
                raise DataJointError(
                    "Could not find database connection in %s" % repr(source[0])
                )

        # initialize graph from dependencies
        connection.dependencies.load()
        super().__init__(connection.dependencies)

        # Enumerate nodes from all the items in the list
        self.nodes_to_show = set()
        try:
            self.nodes_to_show.add(source.full_table_name)
        except AttributeError:
            try:
                database = source.database
            except AttributeError:
                try:
                    database = source.schema.database
                except AttributeError:
                    raise DataJointError(
                        "Cannot plot Diagram for %s" % repr(source)
                    )
            for node in self:
                if node.startswith("`%s`" % database):
                    self.nodes_to_show.add(node)

    @classmethod
    def from_sequence(cls, sequence):
        """
        The join Diagram for all objects in sequence

        :param sequence: a sequence (e.g. list, tuple)
        :return: Diagram(arg1) + ... + Diagram(argn)
        """
        return functools.reduce(lambda x, y: x + y, map(Diagram, sequence))

    def add_parts(self):
        """
        Adds to the diagram the part tables of all master tables already in the diagram
        :return:
        """

        def is_part(part, master):
            """
            :param part:  `database`.`table_name`
            :param master:   `database`.`table_name`
            :return: True if part is part of master.
            """
            part = [s.strip("`") for s in part.split(".")]
            master = [s.strip("`") for s in master.split(".")]
            return (
                master[0] == part[0]
                and master[1] + "__" == part[1][: len(master[1]) + 2]
            )

        self = Diagram(self)  # copy
        self.nodes_to_show.update(
            n
            for n in self.nodes()
            if any(is_part(n, m) for m in self.nodes_to_show)
        )
        return self

    def __add__(self, arg):
        """
        :param arg: either another Diagram or a positive integer.
        :return: Union of the diagrams when arg is another Diagram
                 or an expansion downstream when arg is a positive integer.
        """
        self = Diagram(self)  # copy
        try:
            self.nodes_to_show.update(arg.nodes_to_show)
        except AttributeError:
            try:
                self.nodes_to_show.add(arg.full_table_name)
            except AttributeError:
                for i in range(arg):
                    new = nx.algorithms.boundary.node_boundary(
                        self, self.nodes_to_show
                    )
                    if not new:
                        break
                    # add nodes referenced by aliased nodes
                    new.update(
                        nx.algorithms.boundary.node_boundary(
                            self, (a for a in new if a.isdigit())
                        )
                    )
                    self.nodes_to_show.update(new)
        return self

    def __sub__(self, arg):
        """
        :param arg: either another Diagram or a positive integer.
        :return: Difference of the diagrams when arg is another Diagram or
                 an expansion upstream when arg is a positive integer.
        """
        self = Diagram(self)  # copy
        try:
            self.nodes_to_show.difference_update(arg.nodes_to_show)
        except AttributeError:
            try:
                self.nodes_to_show.remove(arg.full_table_name)
            except AttributeError:
                for i in range(arg):
                    graph = nx.DiGraph(self).reverse()
                    new = nx.algorithms.boundary.node_boundary(
                        graph, self.nodes_to_show
                    )
                    if not new:
                        break
                    # add nodes referenced by aliased nodes
                    new.update(
                        nx.algorithms.boundary.node_boundary(
                            graph, (a for a in new if a.isdigit())
                        )
                    )
                    self.nodes_to_show.update(new)
        return self

    def __mul__(self, arg):
        """
        Intersection of two diagrams
        :param arg: another Diagram
        :return: a new Diagram comprising nodes that are present in both operands.
        """
        self = Diagram(self)  # copy
        self.nodes_to_show.intersection_update(arg.nodes_to_show)
        return self

    def topo_sort(self):
        """return nodes in lexicographical topological order"""
        return topo_sort(self)

    def _make_graph(self):
        """
        Make the self.graph - a graph object ready for drawing
        """
        # mark "distinguished" tables, i.e. those that introduce new primary key
        # attributes
        for name in self.nodes_to_show:
            foreign_attributes = set(
                attr
                for p in self.in_edges(name, data=True)
                for attr in p[2]["attr_map"]
                if p[2]["primary"]
            )
            self.nodes[name]["distinguished"] = (
                "primary_key" in self.nodes[name]
                and foreign_attributes < self.nodes[name]["primary_key"]
            )
        # include aliased nodes that are sandwiched between two displayed nodes
        gaps = set(
            nx.algorithms.boundary.node_boundary(self, self.nodes_to_show)
        ).intersection(
            nx.algorithms.boundary.node_boundary(
                nx.DiGraph(self).reverse(), self.nodes_to_show
            )
        )
        nodes = self.nodes_to_show.union(a for a in gaps if a.isdigit)
        # construct subgraph and rename nodes to class names
        graph = nx.DiGraph(nx.DiGraph(self).subgraph(nodes))
        nx.set_node_attributes(
            graph, name="node_type", values={n: _get_tier(n) for n in graph}
        )
        # relabel nodes to class names
        mapping = {
            node: lookup_class_name(node, self.context) or node
            for node in graph.nodes()
        }
        new_names = [mapping.values()]
        if len(new_names) > len(set(new_names)):
            raise DataJointError(
                "Some classes have identical names. The Diagram cannot be plotted."
            )
        nx.relabel_nodes(graph, mapping, copy=False)
        return graph

    @staticmethod
    def _encapsulate_edge_attributes(graph):
        """
        Modifies the `nx.Graph`'s edge attribute `attr_map` to be a string representation
        of the attribute map, and encapsulates the string in double quotes.
        Changes the graph in place.

        Implements workaround described in
        https://github.com/pydot/pydot/issues/258#issuecomment-795798099
        """
        for u, v, *_, edgedata in graph.edges(data=True):
            if "attr_map" in edgedata:
                graph.edges[u, v]["attr_map"] = '"{0}"'.format(edgedata["attr_map"])

    @staticmethod
    def _encapsulate_node_names(graph):
        """
        Modifies the `nx.Graph`'s node names string representations encapsulated in
        double quotes.
        Changes the graph in place.

        Implements workaround described in
        https://github.com/datajoint/datajoint-python/pull/1176
        """
        nx.relabel_nodes(
            graph,
            {node: '"{0}"'.format(node) for node in graph.nodes()},
            copy=False,
        )

    def make_dot(self):
        graph = self._make_graph()
        graph.nodes()

        scale = 1.2  # scaling factor for fonts and boxes
        label_props = {  # http://matplotlib.org/examples/color/named_colors.html
            None: dict(
                shape="circle",
                color="#FFFF0040",
                fontcolor="yellow",
                fontsize=round(scale * 8),
                size=0.4 * scale,
                fixed=False,
            ),
            _AliasNode: dict(
                shape="circle",
                color="#FF880080",
                fontcolor="#FF880080",
                fontsize=round(scale * 0),
                size=0.05 * scale,
                fixed=True,
            ),
            Manual: dict(
                shape="box",
                color="#00FF0030",
                fontcolor="darkgreen",
                fontsize=round(scale * 10),
                size=0.4 * scale,
                fixed=False,
            ),
            Lookup: dict(
                shape="plaintext",
                color="#00000020",
                fontcolor="black",
                fontsize=round(scale * 8),
                size=0.4 * scale,
                fixed=False,
            ),
            Computed: dict(
                shape="ellipse",
                color="#FF000020",
                fontcolor="#7F0000A0",
                fontsize=round(scale * 10),
                size=0.3 * scale,
                fixed=True,
            ),
            Imported: dict(
                shape="ellipse",
                color="#00007F40",
                fontcolor="#00007FA0",
                fontsize=round(scale * 10),
                size=0.4 * scale,
                fixed=False,
            ),
            Part: dict(
                shape="plaintext",
                color="#0000000",
                fontcolor="black",
                fontsize=round(scale * 8),
                size=0.1 * scale,
                fixed=False,
            ),
        }
        node_props = {
            node: label_props[d["node_type"]]
            for node, d in dict(graph.nodes(data=True)).items()
        }

        self._encapsulate_node_names(graph)
        self._encapsulate_edge_attributes(graph)
        dot = nx.drawing.nx_pydot.to_pydot(graph)
        for node in dot.get_nodes():
            node.set_shape("circle")
            name = node.get_name().strip('"')
            props = node_props[name]
            node.set_fontsize(props["fontsize"])
            node.set_fontcolor(props["fontcolor"])
            node.set_shape(props["shape"])
            node.set_fontname("arial")
            node.set_fixedsize("shape" if props["fixed"] else False)
            node.set_width(props["size"])
            node.set_height(props["size"])
            if name.split(".")[0] in self.context:
                cls = eval(name, self.context)
                assert issubclass(cls, Table)
                description = cls().describe(context=self.context).split("\n")
                description = (
                    (
                        "-" * 30
                        if q.startswith("---")
                        else (
                            q.replace("->", "&#8594;")
                            if "->" in q
                            else q.split(":")[0]
                        )
                    )
                    for q in description
                    if not q.startswith("#")
                )
                node.set_tooltip("&#13;".join(description))
            node.set_label(
                "<<u>" + name + "</u>>"
                if node.get("distinguished") == "True"
                else name
            )
            node.set_color(props["color"])
            node.set_style("filled")

        for edge in dot.get_edges():
            # see https://graphviz.org/doc/info/attrs.html
            src = edge.get_source()
            dest = edge.get_destination()
            props = graph.get_edge_data(src, dest)
            if props is None:
                raise DataJointError(
                    "Could not find edge with source "
                    "'{}' and destination '{}'".format(src, dest)
                )
            edge.set_color("#00000040")
            edge.set_style("solid" if props["primary"] else "dashed")
            master_part = graph.nodes[dest][
                "node_type"
            ] is Part and dest.startswith(src + ".")
            edge.set_weight(3 if master_part else 1)
            edge.set_arrowhead("none")
            edge.set_penwidth(0.75 if props["multi"] else 2)

        return dot

    def make_svg(self):
        from IPython.display import SVG

        return SVG(self.make_dot().create_svg())

    def make_png(self):
        return io.BytesIO(self.make_dot().create_png())

    def make_image(self):
        if plot_active:
            return plt.imread(self.make_png())
        else:
            raise DataJointError("pyplot was not imported")

    def _repr_svg_(self):
        return self.make_svg()._repr_svg_()

    def draw(self):
        if plot_active:
            plt.imshow(self.make_image())
            plt.gca().axis("off")
            plt.show()
        else:
            raise DataJointError("pyplot was not imported")

    def save(self, filename, format=None):
        if format is None:
            if filename.lower().endswith(".png"):
                format = "png"
            elif filename.lower().endswith(".svg"):
                format = "svg"
        if format.lower() == "png":
            with open(filename, "wb") as f:
                f.write(self.make_png().getbuffer().tobytes())
        elif format.lower() == "svg":
            with open(filename, "w") as f:
                f.write(self.make_svg().data)
        else:
            raise DataJointError("Unsupported file format")

    @staticmethod
    def _layout(graph, **kwargs):
        return pydot_layout(graph, prog="dot", **kwargs)

from_sequence(sequence) classmethod

The join Diagram for all objects in sequence

Parameters:

Name	Type	Description	Default
sequence		a sequence (e.g. list, tuple)	required

Returns:

Type	Description
	Diagram(arg1) + ... + Diagram(argn)

Source code in datajoint/diagram.py

@classmethod
def from_sequence(cls, sequence):
    """
    The join Diagram for all objects in sequence

    :param sequence: a sequence (e.g. list, tuple)
    :return: Diagram(arg1) + ... + Diagram(argn)
    """
    return functools.reduce(lambda x, y: x + y, map(Diagram, sequence))

add_parts()

Adds to the diagram the part tables of all master tables already in the diagram

Returns:

Type	Description

Source code in datajoint/diagram.py

def add_parts(self):
    """
    Adds to the diagram the part tables of all master tables already in the diagram
    :return:
    """

    def is_part(part, master):
        """
        :param part:  `database`.`table_name`
        :param master:   `database`.`table_name`
        :return: True if part is part of master.
        """
        part = [s.strip("`") for s in part.split(".")]
        master = [s.strip("`") for s in master.split(".")]
        return (
            master[0] == part[0]
            and master[1] + "__" == part[1][: len(master[1]) + 2]
        )

    self = Diagram(self)  # copy
    self.nodes_to_show.update(
        n
        for n in self.nodes()
        if any(is_part(n, m) for m in self.nodes_to_show)
    )
    return self

topo_sort()

return nodes in lexicographical topological order

Source code in datajoint/diagram.py

def topo_sort(self):
    """return nodes in lexicographical topological order"""
    return topo_sort(self)

kill(restriction=None, connection=None, order_by=None)

view and kill database connections.

Parameters:

Name	Type	Description	Default
restriction		restriction to be applied to processlist	None
connection		a datajoint.Connection object. Default calls datajoint.conn()	None
order_by		order by a single attribute or the list of attributes. defaults to 'id'. Restrictions are specified as strings and can involve any of the attributes of information_schema.processlist: ID, USER, HOST, DB, COMMAND, TIME, STATE, INFO. Examples: dj.kill('HOST LIKE "%compute%"') lists only connections from hosts containing "compute". dj.kill('TIME > 600') lists only connections in their current state for more than 10 minutes	None

Source code in datajoint/admin.py

def kill(restriction=None, connection=None, order_by=None):
    """
    view and kill database connections.

    :param restriction: restriction to be applied to processlist
    :param connection: a datajoint.Connection object. Default calls datajoint.conn()
    :param order_by: order by a single attribute or the list of attributes. defaults to 'id'.

    Restrictions are specified as strings and can involve any of the attributes of
    information_schema.processlist: ID, USER, HOST, DB, COMMAND, TIME, STATE, INFO.

    Examples:
        dj.kill('HOST LIKE "%compute%"') lists only connections from hosts containing "compute".
        dj.kill('TIME > 600') lists only connections in their current state for more than 10 minutes
    """

    if connection is None:
        connection = conn()

    if order_by is not None and not isinstance(order_by, str):
        order_by = ",".join(order_by)

    query = (
        "SELECT * FROM information_schema.processlist WHERE id <> CONNECTION_ID()"
        + ("" if restriction is None else " AND (%s)" % restriction)
        + (" ORDER BY %s" % (order_by or "id"))
    )

    while True:
        print("  ID USER         HOST          STATE         TIME    INFO")
        print("+--+ +----------+ +-----------+ +-----------+ +-----+")
        cur = (
            {k.lower(): v for k, v in elem.items()}
            for elem in connection.query(query, as_dict=True)
        )
        for process in cur:
            try:
                print(
                    "{id:>4d} {user:<12s} {host:<12s} {state:<12s} {time:>7d}  {info}".format(
                        **process
                    )
                )
            except TypeError:
                print(process)
        response = input('process to kill or "q" to quit > ')
        if response == "q":
            break
        if response:
            try:
                pid = int(response)
            except ValueError:
                pass  # ignore non-numeric input
            else:
                try:
                    connection.query("kill %d" % pid)
                except pymysql.err.InternalError:
                    logger.warn("Process not found")

MatCell

Bases: ndarray

a numpy ndarray representing a Matlab cell array

Source code in datajoint/blob.py

class MatCell(np.ndarray):
    """a numpy ndarray representing a Matlab cell array"""

    pass

MatStruct

Bases: recarray

numpy.recarray representing a Matlab struct array

Source code in datajoint/blob.py

class MatStruct(np.recarray):
    """numpy.recarray representing a Matlab struct array"""

    pass

key

object that allows requesting the primary key as an argument in expression.fetch() The string "KEY" can be used instead of the class key

Source code in datajoint/fetch.py

class key:
    """
    object that allows requesting the primary key as an argument in expression.fetch()
    The string "KEY" can be used instead of the class key
    """

    pass

key_hash(mapping)

32-byte hash of the mapping's key values sorted by the key name. This is often used to convert a long primary key value into a shorter hash. For example, the JobTable in datajoint.jobs uses this function to hash the primary key of autopopulated tables.

Source code in datajoint/hash.py

def key_hash(mapping):
    """
    32-byte hash of the mapping's key values sorted by the key name.
    This is often used to convert a long primary key value into a shorter hash.
    For example, the JobTable in datajoint.jobs uses this function to hash the primary key of autopopulated tables.
    """
    hashed = hashlib.md5()
    for k, v in sorted(mapping.items()):
        hashed.update(str(v).encode())
    return hashed.hexdigest()

AttributeAdapter

Base class for adapter objects for user-defined attribute types.

Source code in datajoint/attribute_adapter.py

class AttributeAdapter:
    """
    Base class for adapter objects for user-defined attribute types.
    """

    @property
    def attribute_type(self):
        """
        :return: a supported DataJoint attribute type to use; e.g. "longblob", "blob@store"
        """
        raise NotImplementedError("Undefined attribute adapter")

    def get(self, value):
        """
        convert value retrieved from the the attribute in a table into the adapted type

        :param value: value from the database

        :return: object of the adapted type
        """
        raise NotImplementedError("Undefined attribute adapter")

    def put(self, obj):
        """
        convert an object of the adapted type into a value that DataJoint can store in a table attribute

        :param obj: an object of the adapted type
        :return: value to store in the database
        """
        raise NotImplementedError("Undefined attribute adapter")

attribute_type property

Returns:

Type	Description
	a supported DataJoint attribute type to use; e.g. "longblob", "blob@store"

get(value)

convert value retrieved from the the attribute in a table into the adapted type

Parameters:

Name	Type	Description	Default
value		value from the database	required

Returns:

Type	Description
	object of the adapted type

Source code in datajoint/attribute_adapter.py

def get(self, value):
    """
    convert value retrieved from the the attribute in a table into the adapted type

    :param value: value from the database

    :return: object of the adapted type
    """
    raise NotImplementedError("Undefined attribute adapter")

put(obj)

convert an object of the adapted type into a value that DataJoint can store in a table attribute

Parameters:

Name	Type	Description	Default
obj		an object of the adapted type	required

Returns:

Type	Description
	value to store in the database

Source code in datajoint/attribute_adapter.py

def put(self, obj):
    """
    convert an object of the adapted type into a value that DataJoint can store in a table attribute

    :param obj: an object of the adapted type
    :return: value to store in the database
    """
    raise NotImplementedError("Undefined attribute adapter")

DataJointError

Bases: Exception

Base class for errors specific to DataJoint internal operation.

Source code in datajoint/errors.py

class DataJointError(Exception):
    """
    Base class for errors specific to DataJoint internal operation.
    """

    def __init__(self, *args):
        from .plugin import connection_plugins, type_plugins

        self.__cause__ = (
            PluginWarning("Unverified DataJoint plugin detected.")
            if any(
                [
                    any([not plugins[k]["verified"] for k in plugins])
                    for plugins in [connection_plugins, type_plugins]
                    if plugins
                ]
            )
            else None
        )

    def suggest(self, *args):
        """
        regenerate the exception with additional arguments

        :param args: addition arguments
        :return: a new exception of the same type with the additional arguments
        """
        return self.__class__(*(self.args + args))

suggest(*args)

regenerate the exception with additional arguments

Parameters:

Name	Type	Description	Default
args		addition arguments	()

Returns:

Type	Description
	a new exception of the same type with the additional arguments

Source code in datajoint/errors.py

def suggest(self, *args):
    """
    regenerate the exception with additional arguments

    :param args: addition arguments
    :return: a new exception of the same type with the additional arguments
    """
    return self.__class__(*(self.args + args))