#11958

pending completion

Build # #11958

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

<a href="https://github.com/datajoint/datajoint-python/commit/<a class=hub.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://git"><a class=hub.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d0f4">ce766e8cd<a href="https://github.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d0f4">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/datajoint/datajoint-python/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d0f4&quot;&gt;ce766e8cd&lt;/a&gt;&lt;a href=&quot;https://github.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d0f4&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3b&lt;/a&gt;c966d0f4&quot;>&quot;&gt;Merge &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/datajoint/datajoint-python/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/c5f0a193c6a396c058e9fee25ab135ded11faeab">c5f0a193c</a><a href="https://github.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d0f4">&amp;quot;&amp;gt;c5f0a193c&amp;lt;/a&amp;gt;&amp;quot;&amp;gt;c5f0a193c&amp;lt;/a&amp;gt;&amp;lt;a href=&amp;quot;https://github.com/datajoint/datajoint-python/commit/ce766e8cd4b243488c1be2cea0302f3bc966d&lt;/a&gt;0f4&quot;&gt; into &lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/413e2f90d">413e2f90d">413e2f90d</a>

Run Details

801 of 801 new or added lines in 25 files covered. (100.0%)

3061 of 3386 relevant lines covered (90.4%)

0.9 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.53

/datajoint/heading.py

import numpy as np
from collections import namedtuple, defaultdict
from itertools import chain
import re
import logging
from .errors import DataJointError, _support_filepath_types, FILEPATH_FEATURE_SWITCH
from .declare import (
    UUID_DATA_TYPE,
    SPECIAL_TYPES,
    TYPE_PATTERN,
    EXTERNAL_TYPES,
    NATIVE_TYPES,
)
from .attribute_adapter import get_adapter, AttributeAdapter


logger = logging.getLogger(__name__)

default_attribute_properties = (
    dict(  # these default values are set in computed attributes
        name=None,
        type="expression",
        in_key=False,
        nullable=False,
        default=None,
        comment="calculated attribute",
        autoincrement=False,
        numeric=None,
        string=None,
        uuid=False,
        is_blob=False,
        is_attachment=False,
        is_filepath=False,
        is_external=False,
        adapter=None,
        store=None,
        unsupported=False,
        attribute_expression=None,
        database=None,
        dtype=object,
    )
)


class Attribute(namedtuple("_Attribute", default_attribute_properties)):
    """
    Properties of a table column (attribute)
    """

    def todict(self):
        """Convert namedtuple to dict."""
        return dict((name, self[i]) for i, name in enumerate(self._fields))

    @property
    def sql_type(self):
        """:return: datatype (as string) in database. In most cases, it is the same as self.type"""
        return UUID_DATA_TYPE if self.uuid else self.type

    @property
    def sql_comment(self):
        """:return: full comment for the SQL declaration. Includes custom type specification"""
        return (":uuid:" if self.uuid else "") + self.comment

    @property
    def sql(self):
        """
        Convert primary key attribute tuple into its SQL CREATE TABLE clause.
        Default values are not reflected.
        This is used for declaring foreign keys in referencing tables
        :return: SQL code for attribute declaration
        """
        return '`{name}` {type} NOT NULL COMMENT "{comment}"'.format(
            name=self.name, type=self.sql_type, comment=self.sql_comment
        )

    @property
    def original_name(self):
        if self.attribute_expression is None:
            return self.name
        assert self.attribute_expression.startswith("`")
        return self.attribute_expression.strip("`")


class Heading:
    """
    Local class for relations' headings.
    Heading contains the property attributes, which is an dict in which the keys are
    the attribute names and the values are Attributes.
    """

    def __init__(self, attribute_specs=None, table_info=None):
        """
        :param attribute_specs: a list of dicts with the same keys as Attribute
        :param table_info: a dict with information to load the heading from the database
        """
        self.indexes = None
        self.table_info = table_info
        self._table_status = None
        self._attributes = (
            None
            if attribute_specs is None
            else dict((q["name"], Attribute(**q)) for q in attribute_specs)
        )

    def __len__(self):
        return 0 if self.attributes is None else len(self.attributes)

    @property
    def table_status(self):
        if self.table_info is None:
            return None
        if self._table_status is None:
            self._init_from_database()
        return self._table_status

    @property
    def attributes(self):
        if self._attributes is None:
            self._init_from_database()  # lazy loading from database
        return self._attributes

    @property
    def names(self):
        return [k for k in self.attributes]

    @property
    def primary_key(self):
        return [k for k, v in self.attributes.items() if v.in_key]

    @property
    def secondary_attributes(self):
        return [k for k, v in self.attributes.items() if not v.in_key]

    @property
    def blobs(self):
        return [k for k, v in self.attributes.items() if v.is_blob]

    @property
    def non_blobs(self):
        return [
            k
            for k, v in self.attributes.items()
            if not v.is_blob and not v.is_attachment and not v.is_filepath
        ]

    @property
    def new_attributes(self):
        return [
            k for k, v in self.attributes.items() if v.attribute_expression is not None
        ]

    def __getitem__(self, name):
        """shortcut to the attribute"""
        return self.attributes[name]

    def __repr__(self):
        """
        :return:  heading representation in DataJoint declaration format but without foreign key expansion
        """
        in_key = True
        ret = ""
        if self._table_status is not None:
            ret += "# " + self.table_status["comment"] + "\n"
        for v in self.attributes.values():
            if in_key and not v.in_key:
                ret += "---\n"
                in_key = False
            ret += "%-20s : %-28s # %s\n" % (
                v.name if v.default is None else "%s=%s" % (v.name, v.default),
                "%s%s" % (v.type, "auto_increment" if v.autoincrement else ""),
                v.comment,
            )
        return ret

    @property
    def has_autoincrement(self):
        return any(e.autoincrement for e in self.attributes.values())

    @property
    def as_dtype(self):
        """
        represent the heading as a numpy dtype
        """
        return np.dtype(
            dict(names=self.names, formats=[v.dtype for v in self.attributes.values()])
        )

    def as_sql(self, fields, include_aliases=True):
        """
        represent heading as the SQL SELECT clause.
        """
        return ",".join(
            "`%s`" % name
            if self.attributes[name].attribute_expression is None
            else self.attributes[name].attribute_expression
            + (" as `%s`" % name if include_aliases else "")
            for name in fields
        )

    def __iter__(self):
        return iter(self.attributes)

    def _init_from_database(self):
        """initialize heading from an existing database table."""
        conn, database, table_name, context = (
            self.table_info[k] for k in ("conn", "database", "table_name", "context")
        )
        info = conn.query(
            'SHOW TABLE STATUS FROM `{database}` WHERE name="{table_name}"'.format(
                table_name=table_name, database=database
            ),
            as_dict=True,
        ).fetchone()
        if info is None:
            if table_name == "~log":
                logger.warning("Could not create the ~log table")
                return
            raise DataJointError(
                "The table `{database}`.`{table_name}` is not defined.".format(
                    table_name=table_name, database=database
                )
            )
        self._table_status = {k.lower(): v for k, v in info.items()}
        cur = conn.query(
            "SHOW FULL COLUMNS FROM `{table_name}` IN `{database}`".format(
                table_name=table_name, database=database
            ),
            as_dict=True,
        )

        attributes = cur.fetchall()

        rename_map = {
            "Field": "name",
            "Type": "type",
            "Null": "nullable",
            "Default": "default",
            "Key": "in_key",
            "Comment": "comment",
        }

        fields_to_drop = ("Privileges", "Collation")

        # rename and drop attributes
        attributes = [
            {
                rename_map[k] if k in rename_map else k: v
                for k, v in x.items()
                if k not in fields_to_drop
            }
            for x in attributes
        ]
        numeric_types = {
            ("float", False): np.float64,
            ("float", True): np.float64,
            ("double", False): np.float64,
            ("double", True): np.float64,
            ("tinyint", False): np.int64,
            ("tinyint", True): np.int64,
            ("smallint", False): np.int64,
            ("smallint", True): np.int64,
            ("mediumint", False): np.int64,
            ("mediumint", True): np.int64,
            ("int", False): np.int64,
            ("int", True): np.int64,
            ("bigint", False): np.int64,
            ("bigint", True): np.uint64,
        }

        sql_literals = ["CURRENT_TIMESTAMP"]

        # additional attribute properties
        for attr in attributes:

            attr.update(
                in_key=(attr["in_key"] == "PRI"),
                database=database,
                nullable=attr["nullable"] == "YES",
                autoincrement=bool(
                    re.search(r"auto_increment", attr["Extra"], flags=re.I)
                ),
                numeric=any(
                    TYPE_PATTERN[t].match(attr["type"])
                    for t in ("DECIMAL", "INTEGER", "FLOAT")
                ),
                string=any(
                    TYPE_PATTERN[t].match(attr["type"])
                    for t in ("ENUM", "TEMPORAL", "STRING")
                ),
                is_blob=bool(TYPE_PATTERN["INTERNAL_BLOB"].match(attr["type"])),
                uuid=False,
                is_attachment=False,
                is_filepath=False,
                adapter=None,
                store=None,
                is_external=False,
                attribute_expression=None,
            )

            if any(TYPE_PATTERN[t].match(attr["type"]) for t in ("INTEGER", "FLOAT")):
                attr["type"] = re.sub(
                    r"\(\d+\)", "", attr["type"], count=1
                )  # strip size off integers and floats
            attr["unsupported"] = not any(
                (attr["is_blob"], attr["numeric"], attr["numeric"])
            )
            attr.pop("Extra")

            # process custom DataJoint types
            special = re.match(r":(?P<type>[^:]+):(?P<comment>.*)", attr["comment"])
            if special:
                special = special.groupdict()
                attr.update(special)
            # process adapted attribute types
            if special and TYPE_PATTERN["ADAPTED"].match(attr["type"]):
                assert context is not None, "Declaration context is not set"
                adapter_name = special["type"]
                try:
                    attr.update(adapter=get_adapter(context, adapter_name))
                except DataJointError:
                    # if no adapter, then delay the error until the first invocation
                    attr.update(adapter=AttributeAdapter())
                else:
                    attr.update(type=attr["adapter"].attribute_type)
                    if not any(r.match(attr["type"]) for r in TYPE_PATTERN.values()):
                        raise DataJointError(
                            "Invalid attribute type '{type}' in adapter object <{adapter_name}>.".format(
                                adapter_name=adapter_name, **attr
                            )
                        )
                    special = not any(
                        TYPE_PATTERN[c].match(attr["type"]) for c in NATIVE_TYPES
                    )

            if special:
                try:
                    category = next(
                        c for c in SPECIAL_TYPES if TYPE_PATTERN[c].match(attr["type"])
                    )
                except StopIteration:
                    if attr["type"].startswith("external"):
                        url = (
                            "https://docs.datajoint.io/python/admin/5-blob-config.html"
                            "#migration-between-datajoint-v0-11-and-v0-12"
                        )
                        raise DataJointError(
                            "Legacy datatype `{type}`. Migrate your external stores to "
                            "datajoint 0.12: {url}".format(url=url, **attr)
                        )
                    raise DataJointError(
                        "Unknown attribute type `{type}`".format(**attr)
                    )
                if category == "FILEPATH" and not _support_filepath_types():
                    raise DataJointError(
                        """
                        The filepath data type is disabled until complete validation.
                        To turn it on as experimental feature, set the environment variable
                        {env} = TRUE or upgrade datajoint.
                        """.format(
                            env=FILEPATH_FEATURE_SWITCH
                        )
                    )
                attr.update(
                    unsupported=False,
                    is_attachment=category in ("INTERNAL_ATTACH", "EXTERNAL_ATTACH"),
                    is_filepath=category == "FILEPATH",
                    # INTERNAL_BLOB is not a custom type but is included for completeness
                    is_blob=category in ("INTERNAL_BLOB", "EXTERNAL_BLOB"),
                    uuid=category == "UUID",
                    is_external=category in EXTERNAL_TYPES,
                    store=attr["type"].split("@")[1]
                    if category in EXTERNAL_TYPES
                    else None,
                )

            if attr["in_key"] and any(
                (attr["is_blob"], attr["is_attachment"], attr["is_filepath"])
            ):
                raise DataJointError(
                    "Blob, attachment, or filepath attributes are not allowed in the primary key"
                )

            if (
                attr["string"]
                and attr["default"] is not None
                and attr["default"] not in sql_literals
            ):
                attr["default"] = '"%s"' % attr["default"]

            if attr["nullable"]:  # nullable fields always default to null
                attr["default"] = "null"

            # fill out dtype. All floats and non-nullable integers are turned into specific dtypes
            attr["dtype"] = object
            if attr["numeric"] and not attr["adapter"]:
                is_integer = TYPE_PATTERN["INTEGER"].match(attr["type"])
                is_float = TYPE_PATTERN["FLOAT"].match(attr["type"])
                if is_integer and not attr["nullable"] or is_float:
                    is_unsigned = bool(re.match("sunsigned", attr["type"], flags=re.I))
                    t = re.sub(r"\(.*\)", "", attr["type"])  # remove parentheses
                    t = re.sub(r" unsigned$", "", t)  # remove unsigned
                    assert (t, is_unsigned) in numeric_types, (
                        "dtype not found for type %s" % t
                    )
                    attr["dtype"] = numeric_types[(t, is_unsigned)]

            if attr["adapter"]:
                # restore adapted type name
                attr["type"] = adapter_name

        self._attributes = dict(((q["name"], Attribute(**q)) for q in attributes))

        # Read and tabulate secondary indexes
        keys = defaultdict(dict)
        for item in conn.query(
            "SHOW KEYS FROM `{db}`.`{tab}`".format(db=database, tab=table_name),
            as_dict=True,
        ):
            if item["Key_name"] != "PRIMARY":
                keys[item["Key_name"]][item["Seq_in_index"]] = dict(
                    column=item["Column_name"],
                    unique=(item["Non_unique"] == 0),
                    nullable=item["Null"].lower() == "yes",
                )
        self.indexes = {
            tuple(item[k]["column"] for k in sorted(item.keys())): dict(
                unique=item[1]["unique"],
                nullable=any(v["nullable"] for v in item.values()),
            )
            for item in keys.values()
        }

    def select(self, select_list, rename_map=None, compute_map=None):
        """
        derive a new heading by selecting, renaming, or computing attributes.
        In relational algebra these operators are known as project, rename, and extend.
        :param select_list:  the full list of existing attributes to include
        :param rename_map:  dictionary of renamed attributes: keys=new names, values=old names
        :param compute_map: a direction of computed attributes
        This low-level method performs no error checking.
        """
        rename_map = rename_map or {}
        compute_map = compute_map or {}
        copy_attrs = list()
        for name in self.attributes:
            if name in select_list:
                copy_attrs.append(self.attributes[name].todict())
            copy_attrs.extend(
                (
                    dict(
                        self.attributes[old_name].todict(),
                        name=new_name,
                        attribute_expression="`%s`" % old_name,
                    )
                    for new_name, old_name in rename_map.items()
                    if old_name == name
                )
            )
        compute_attrs = (
            dict(default_attribute_properties, name=new_name, attribute_expression=expr)
            for new_name, expr in compute_map.items()
        )
        return Heading(chain(copy_attrs, compute_attrs))

    def join(self, other):
        """
        Join two headings into a new one.
        It assumes that self and other are headings that share no common dependent attributes.
        """
        return Heading(
            [self.attributes[name].todict() for name in self.primary_key]
            + [
                other.attributes[name].todict()
                for name in other.primary_key
                if name not in self.primary_key
            ]
            + [
                self.attributes[name].todict()
                for name in self.secondary_attributes
                if name not in other.primary_key
            ]
            + [
                other.attributes[name].todict()
                for name in other.secondary_attributes
                if name not in self.primary_key
            ]
        )

    def set_primary_key(self, primary_key):
        """
        Create a new heading with the specified primary key.
        This low-level method performs no error checking.
        """
        return Heading(
            chain(
                (
                    dict(self.attributes[name].todict(), in_key=True)
                    for name in primary_key
                ),
                (
                    dict(self.attributes[name].todict(), in_key=False)
                    for name in self.names
                    if name not in primary_key
                ),
            )
        )

    def make_subquery_heading(self):
        """
        Create a new heading with removed attribute sql_expressions.
        Used by subqueries, which resolve the sql_expressions.
        """
        return Heading(
            dict(v.todict(), attribute_expression=None)
            for v in self.attributes.values()
        )

1	import numpy as np	1✔
2	from collections import namedtuple, defaultdict	1✔
3	from itertools import chain	1✔
4	import re	1✔
5	import logging	1✔
6	from .errors import DataJointError, _support_filepath_types, FILEPATH_FEATURE_SWITCH	1✔
7	from .declare import (	1✔
8	UUID_DATA_TYPE,
9	SPECIAL_TYPES,
10	TYPE_PATTERN,
11	EXTERNAL_TYPES,
12	NATIVE_TYPES,
13	)
14	from .attribute_adapter import get_adapter, AttributeAdapter	1✔
15
16
17	logger = logging.getLogger(__name__)	1✔
18
19	default_attribute_properties = (	1✔
20	dict( # these default values are set in computed attributes
21	name=None,
22	type="expression",
23	in_key=False,
24	nullable=False,
25	default=None,
26	comment="calculated attribute",
27	autoincrement=False,
28	numeric=None,
29	string=None,
30	uuid=False,
31	is_blob=False,
32	is_attachment=False,
33	is_filepath=False,
34	is_external=False,
35	adapter=None,
36	store=None,
37	unsupported=False,
38	attribute_expression=None,
39	database=None,
40	dtype=object,
41	)
42	)
43
44
45	class Attribute(namedtuple("_Attribute", default_attribute_properties)):	1✔
46	"""
47	Properties of a table column (attribute)
48	"""
49
50	def todict(self):	1✔
51	"""Convert namedtuple to dict."""
52	return dict((name, self[i]) for i, name in enumerate(self._fields))	1✔
53
54	@property	1✔
55	def sql_type(self):	1✔
56	""":return: datatype (as string) in database. In most cases, it is the same as self.type"""
57	return UUID_DATA_TYPE if self.uuid else self.type	1✔
58
59	@property	1✔
60	def sql_comment(self):	1✔
61	""":return: full comment for the SQL declaration. Includes custom type specification"""
62	return (":uuid:" if self.uuid else "") + self.comment	1✔
63
64	@property	1✔
65	def sql(self):	1✔
66	"""
67	Convert primary key attribute tuple into its SQL CREATE TABLE clause.
68	Default values are not reflected.
69	This is used for declaring foreign keys in referencing tables
70	:return: SQL code for attribute declaration
71	"""
72	return '`{name}` {type} NOT NULL COMMENT "{comment}"'.format(	1✔
73	name=self.name, type=self.sql_type, comment=self.sql_comment
74	)
75
76	@property	1✔
77	def original_name(self):	1✔
78	if self.attribute_expression is None:	1✔
79	return self.name	1✔
80	assert self.attribute_expression.startswith("`")	1✔
81	return self.attribute_expression.strip("`")	1✔
82
83
84	class Heading:	1✔
85	"""
86	Local class for relations' headings.
87	Heading contains the property attributes, which is an dict in which the keys are
88	the attribute names and the values are Attributes.
89	"""
90
91	def __init__(self, attribute_specs=None, table_info=None):	1✔
92	"""
93	:param attribute_specs: a list of dicts with the same keys as Attribute
94	:param table_info: a dict with information to load the heading from the database
95	"""
96	self.indexes = None	1✔
97	self.table_info = table_info	1✔
98	self._table_status = None	1✔
99	self._attributes = (	1✔
100	None
101	if attribute_specs is None
102	else dict((q["name"], Attribute(**q)) for q in attribute_specs)
103	)
104
105	def __len__(self):	1✔
106	return 0 if self.attributes is None else len(self.attributes)	1✔
107
108	@property	1✔
109	def table_status(self):	1✔
110	if self.table_info is None:	1✔
111	return None	×
112	if self._table_status is None:	1✔
113	self._init_from_database()	1✔
114	return self._table_status	1✔
115
116	@property	1✔
117	def attributes(self):	1✔
118	if self._attributes is None:	1✔
119	self._init_from_database() # lazy loading from database	1✔
120	return self._attributes	1✔
121
122	@property	1✔
123	def names(self):	1✔
124	return [k for k in self.attributes]	1✔
125
126	@property	1✔
127	def primary_key(self):	1✔
128	return [k for k, v in self.attributes.items() if v.in_key]	1✔
129
130	@property	1✔
131	def secondary_attributes(self):	1✔
132	return [k for k, v in self.attributes.items() if not v.in_key]	1✔
133
134	@property	1✔
135	def blobs(self):	1✔
136	return [k for k, v in self.attributes.items() if v.is_blob]	×
137
138	@property	1✔
139	def non_blobs(self):	1✔
140	return [	1✔
141	k
142	for k, v in self.attributes.items()
143	if not v.is_blob and not v.is_attachment and not v.is_filepath
144	]
145
146	@property	1✔
147	def new_attributes(self):	1✔
148	return [	1✔
149	k for k, v in self.attributes.items() if v.attribute_expression is not None
150	]
151
152	def __getitem__(self, name):	1✔
153	"""shortcut to the attribute"""
154	return self.attributes[name]	1✔
155
156	def __repr__(self):	1✔
157	"""
158	:return: heading representation in DataJoint declaration format but without foreign key expansion
159	"""
160	in_key = True	1✔
161	ret = ""	1✔
162	if self._table_status is not None:	1✔
163	ret += "# " + self.table_status["comment"] + "\n"	×
164	for v in self.attributes.values():	1✔
165	if in_key and not v.in_key:	1✔
166	ret += "---\n"	1✔
167	in_key = False	1✔
168	ret += "%-20s : %-28s # %s\n" % (	1✔
169	v.name if v.default is None else "%s=%s" % (v.name, v.default),
170	"%s%s" % (v.type, "auto_increment" if v.autoincrement else ""),
171	v.comment,
172	)
173	return ret	1✔
174
175	@property	1✔
176	def has_autoincrement(self):	1✔
177	return any(e.autoincrement for e in self.attributes.values())	1✔
178
179	@property	1✔
180	def as_dtype(self):	1✔
181	"""
182	represent the heading as a numpy dtype
183	"""
184	return np.dtype(	1✔
185	dict(names=self.names, formats=[v.dtype for v in self.attributes.values()])
186	)
187
188	def as_sql(self, fields, include_aliases=True):	1✔
189	"""
190	represent heading as the SQL SELECT clause.
191	"""
192	return ",".join(	1✔
193	"`%s`" % name
194	if self.attributes[name].attribute_expression is None
195	else self.attributes[name].attribute_expression
196	+ (" as `%s`" % name if include_aliases else "")
197	for name in fields
198	)
199
200	def __iter__(self):	1✔
201	return iter(self.attributes)	1✔
202
203	def _init_from_database(self):	1✔
204	"""initialize heading from an existing database table."""
205	conn, database, table_name, context = (	1✔
206	self.table_info[k] for k in ("conn", "database", "table_name", "context")
207	)
208	info = conn.query(	1✔
209	'SHOW TABLE STATUS FROM `{database}` WHERE name="{table_name}"'.format(
210	table_name=table_name, database=database
211	),
212	as_dict=True,
213	).fetchone()
214	if info is None:	1✔
215	if table_name == "~log":	1✔
216	logger.warning("Could not create the ~log table")	×
217	return	×
218	raise DataJointError(	1✔
219	"The table `{database}`.`{table_name}` is not defined.".format(
220	table_name=table_name, database=database
221	)
222	)
223	self._table_status = {k.lower(): v for k, v in info.items()}	1✔
224	cur = conn.query(	1✔
225	"SHOW FULL COLUMNS FROM `{table_name}` IN `{database}`".format(
226	table_name=table_name, database=database
227	),
228	as_dict=True,
229	)
230
231	attributes = cur.fetchall()	1✔
232
233	rename_map = {	1✔
234	"Field": "name",
235	"Type": "type",
236	"Null": "nullable",
237	"Default": "default",
238	"Key": "in_key",
239	"Comment": "comment",
240	}
241
242	fields_to_drop = ("Privileges", "Collation")	1✔
243
244	# rename and drop attributes
245	attributes = [	1✔
246	{
247	rename_map[k] if k in rename_map else k: v
248	for k, v in x.items()
249	if k not in fields_to_drop
250	}
251	for x in attributes
252	]
253	numeric_types = {	1✔
254	("float", False): np.float64,
255	("float", True): np.float64,
256	("double", False): np.float64,
257	("double", True): np.float64,
258	("tinyint", False): np.int64,
259	("tinyint", True): np.int64,
260	("smallint", False): np.int64,
261	("smallint", True): np.int64,
262	("mediumint", False): np.int64,
263	("mediumint", True): np.int64,
264	("int", False): np.int64,
265	("int", True): np.int64,
266	("bigint", False): np.int64,
267	("bigint", True): np.uint64,
268	}
269
270	sql_literals = ["CURRENT_TIMESTAMP"]	1✔
271
272	# additional attribute properties
273	for attr in attributes:	1✔
274
275	attr.update(	1✔
276	in_key=(attr["in_key"] == "PRI"),
277	database=database,
278	nullable=attr["nullable"] == "YES",
279	autoincrement=bool(
280	re.search(r"auto_increment", attr["Extra"], flags=re.I)
281	),
282	numeric=any(
283	TYPE_PATTERN[t].match(attr["type"])
284	for t in ("DECIMAL", "INTEGER", "FLOAT")
285	),
286	string=any(
287	TYPE_PATTERN[t].match(attr["type"])
288	for t in ("ENUM", "TEMPORAL", "STRING")
289	),
290	is_blob=bool(TYPE_PATTERN["INTERNAL_BLOB"].match(attr["type"])),
291	uuid=False,
292	is_attachment=False,
293	is_filepath=False,
294	adapter=None,
295	store=None,
296	is_external=False,
297	attribute_expression=None,
298	)
299
300	if any(TYPE_PATTERN[t].match(attr["type"]) for t in ("INTEGER", "FLOAT")):	1✔
301	attr["type"] = re.sub(	1✔
302	r"\(\d+\)", "", attr["type"], count=1
303	) # strip size off integers and floats
304	attr["unsupported"] = not any(	1✔
305	(attr["is_blob"], attr["numeric"], attr["numeric"])
306	)
307	attr.pop("Extra")	1✔
308
309	# process custom DataJoint types
310	special = re.match(r":(?P<type>[^:]+):(?P<comment>.*)", attr["comment"])	1✔
311	if special:	1✔
312	special = special.groupdict()	1✔
313	attr.update(special)	1✔
314	# process adapted attribute types
315	if special and TYPE_PATTERN["ADAPTED"].match(attr["type"]):	1✔
316	assert context is not None, "Declaration context is not set"	1✔
317	adapter_name = special["type"]	1✔
318	try:	1✔
319	attr.update(adapter=get_adapter(context, adapter_name))	1✔
320	except DataJointError:	×
321	# if no adapter, then delay the error until the first invocation
322	attr.update(adapter=AttributeAdapter())	×
323	else:
324	attr.update(type=attr["adapter"].attribute_type)	1✔
325	if not any(r.match(attr["type"]) for r in TYPE_PATTERN.values()):	1✔
326	raise DataJointError(	×
327	"Invalid attribute type '{type}' in adapter object <{adapter_name}>.".format(
328	adapter_name=adapter_name, **attr
329	)
330	)
331	special = not any(	1✔
332	TYPE_PATTERN[c].match(attr["type"]) for c in NATIVE_TYPES
333	)
334
335	if special:	1✔
336	try:	1✔
337	category = next(	1✔
338	c for c in SPECIAL_TYPES if TYPE_PATTERN[c].match(attr["type"])
339	)
340	except StopIteration:	×
341	if attr["type"].startswith("external"):	×
342	url = (	×
343	"https://docs.datajoint.io/python/admin/5-blob-config.html"
344	"#migration-between-datajoint-v0-11-and-v0-12"
345	)
346	raise DataJointError(	×
347	"Legacy datatype `{type}`. Migrate your external stores to "
348	"datajoint 0.12: {url}".format(url=url, **attr)
349	)
350	raise DataJointError(	×
351	"Unknown attribute type `{type}`".format(**attr)
352	)
353	if category == "FILEPATH" and not _support_filepath_types():	1✔
354	raise DataJointError(	×
355	"""
356	The filepath data type is disabled until complete validation.
357	To turn it on as experimental feature, set the environment variable
358	{env} = TRUE or upgrade datajoint.
359	""".format(
360	env=FILEPATH_FEATURE_SWITCH
361	)
362	)
363	attr.update(	1✔
364	unsupported=False,
365	is_attachment=category in ("INTERNAL_ATTACH", "EXTERNAL_ATTACH"),
366	is_filepath=category == "FILEPATH",
367	# INTERNAL_BLOB is not a custom type but is included for completeness
368	is_blob=category in ("INTERNAL_BLOB", "EXTERNAL_BLOB"),
369	uuid=category == "UUID",
370	is_external=category in EXTERNAL_TYPES,
371	store=attr["type"].split("@")[1]
372	if category in EXTERNAL_TYPES
373	else None,
374	)
375
376	if attr["in_key"] and any(	1✔
377	(attr["is_blob"], attr["is_attachment"], attr["is_filepath"])
378	):
379	raise DataJointError(	×
380	"Blob, attachment, or filepath attributes are not allowed in the primary key"
381	)
382
383	if (	1✔
384	attr["string"]
385	and attr["default"] is not None
386	and attr["default"] not in sql_literals
387	):
388	attr["default"] = '"%s"' % attr["default"]	1✔
389
390	if attr["nullable"]: # nullable fields always default to null	1✔
391	attr["default"] = "null"	1✔
392
393	# fill out dtype. All floats and non-nullable integers are turned into specific dtypes
394	attr["dtype"] = object	1✔
395	if attr["numeric"] and not attr["adapter"]:	1✔
396	is_integer = TYPE_PATTERN["INTEGER"].match(attr["type"])	1✔
397	is_float = TYPE_PATTERN["FLOAT"].match(attr["type"])	1✔
398	if is_integer and not attr["nullable"] or is_float:	1✔
399	is_unsigned = bool(re.match("sunsigned", attr["type"], flags=re.I))	1✔
400	t = re.sub(r"\(.*\)", "", attr["type"]) # remove parentheses	1✔
401	t = re.sub(r" unsigned$", "", t) # remove unsigned	1✔
402	assert (t, is_unsigned) in numeric_types, (	1✔
403	"dtype not found for type %s" % t
404	)
405	attr["dtype"] = numeric_types[(t, is_unsigned)]	1✔
406
407	if attr["adapter"]:	1✔
408	# restore adapted type name
409	attr["type"] = adapter_name	1✔
410
411	self._attributes = dict(((q["name"], Attribute(**q)) for q in attributes))	1✔
412
413	# Read and tabulate secondary indexes
414	keys = defaultdict(dict)	1✔
415	for item in conn.query(	1✔
416	"SHOW KEYS FROM `{db}`.`{tab}`".format(db=database, tab=table_name),
417	as_dict=True,
418	):
419	if item["Key_name"] != "PRIMARY":	1✔
420	keys[item["Key_name"]][item["Seq_in_index"]] = dict(	1✔
421	column=item["Column_name"],
422	unique=(item["Non_unique"] == 0),
423	nullable=item["Null"].lower() == "yes",
424	)
425	self.indexes = {	1✔
426	tuple(item[k]["column"] for k in sorted(item.keys())): dict(
427	unique=item[1]["unique"],
428	nullable=any(v["nullable"] for v in item.values()),
429	)
430	for item in keys.values()
431	}
432
433	def select(self, select_list, rename_map=None, compute_map=None):	1✔
434	"""
435	derive a new heading by selecting, renaming, or computing attributes.
436	In relational algebra these operators are known as project, rename, and extend.
437	:param select_list: the full list of existing attributes to include
438	:param rename_map: dictionary of renamed attributes: keys=new names, values=old names
439	:param compute_map: a direction of computed attributes
440	This low-level method performs no error checking.
441	"""
442	rename_map = rename_map or {}	1✔
443	compute_map = compute_map or {}	1✔
444	copy_attrs = list()	1✔
445	for name in self.attributes:	1✔
446	if name in select_list:	1✔
447	copy_attrs.append(self.attributes[name].todict())	1✔
448	copy_attrs.extend(	1✔
449	(
450	dict(
451	self.attributes[old_name].todict(),
452	name=new_name,
453	attribute_expression="`%s`" % old_name,
454	)
455	for new_name, old_name in rename_map.items()
456	if old_name == name
457	)
458	)
459	compute_attrs = (	1✔
460	dict(default_attribute_properties, name=new_name, attribute_expression=expr)
461	for new_name, expr in compute_map.items()
462	)
463	return Heading(chain(copy_attrs, compute_attrs))	1✔
464
465	def join(self, other):	1✔
466	"""
467	Join two headings into a new one.
468	It assumes that self and other are headings that share no common dependent attributes.
469	"""
470	return Heading(	1✔
471	[self.attributes[name].todict() for name in self.primary_key]
472	+ [
473	other.attributes[name].todict()
474	for name in other.primary_key
475	if name not in self.primary_key
476	]
477	+ [
478	self.attributes[name].todict()
479	for name in self.secondary_attributes
480	if name not in other.primary_key
481	]
482	+ [
483	other.attributes[name].todict()
484	for name in other.secondary_attributes
485	if name not in self.primary_key
486	]
487	)
488
489	def set_primary_key(self, primary_key):	1✔
490	"""
491	Create a new heading with the specified primary key.
492	This low-level method performs no error checking.
493	"""
494	return Heading(	1✔
495	chain(
496	(
497	dict(self.attributes[name].todict(), in_key=True)
498	for name in primary_key
499	),
500	(
501	dict(self.attributes[name].todict(), in_key=False)
502	for name in self.names
503	if name not in primary_key
504	),
505	)
506	)
507
508	def make_subquery_heading(self):	1✔
509	"""
510	Create a new heading with removed attribute sql_expressions.
511	Used by subqueries, which resolve the sql_expressions.
512	"""
513	return Heading(	1✔
514	dict(v.todict(), attribute_expression=None)
515	for v in self.attributes.values()
516	)

datajoint / datajoint-python / #11958

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous