#12748

pending completion

Build # #12748

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/b1103fb91ae50b86159eb20dca980d725c1e425a">b1103fb91<a href="https://github.com/datajoint/datajoint-python/commit/b1103fb91ae50b86159eb20dca980d725c1e425a">">Merge </a><a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/<a class="double-link" href="https://github.com/datajoint/datajoint-python/commit/0a239dc38b80a6e9b904e43b86ef828341a04316">0a239dc38</a>">0a239dc38</a><a href="https://github.com/datajoint/datajoint-python/commit/b1103fb91ae50b86159eb20dca980d725c1e425a"> into 03dfbaf8a">03dfbaf8a</a>

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

3020 of 3343 relevant lines covered (90.34%)

0.9 hits per line

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

90.8

/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__.split(".")[0])

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

datajoint / datajoint-python / #12748

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