#12900

pending completion

Build # #12900

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/864be0ccca479b08973e3dc4531e096bf97088fa">864be0ccc<a href="https://github.com/datajoint/datajoint-python/commit/864be0ccca479b08973e3dc4531e096bf97088fa">">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/d3b1af13150e5e3a26410b98f2dc2a19ec2b5368">d3b1af131</a>">d3b1af131</a><a href="https://github.com/datajoint/datajoint-python/commit/864be0ccca479b08973e3dc4531e096bf97088fa"> into 3b6e84588">3b6e84588</a>

Run Details

81 of 81 new or added lines in 10 files covered. (100.0%)

3134 of 3489 relevant lines covered (89.83%)

0.9 hits per line

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

90.91

/datajoint/utils.py

"""General-purpose utilities"""

import hashlib
import re
from pathlib import Path
import shutil
import uuid
from .errors import DataJointError
from .logging import logger


class ClassProperty:
    def __init__(self, f):
        self.f = f

    def __get__(self, obj, owner):
        return self.f(owner)


def user_choice(prompt, choices=("yes", "no"), default=None):
    """
    Prompts the user for confirmation.  The default value, if any, is capitalized.

    :param prompt: Information to display to the user.
    :param choices: an iterable of possible choices.
    :param default: default choice
    :return: the user's choice
    """
    assert default is None or default in choices
    choice_list = ", ".join(
        (choice.title() if choice == default else choice for choice in choices)
    )
    response = None
    while response not in choices:
        response = input(prompt + " [" + choice_list + "]: ")
        response = response.lower() if response else default
    return response


def get_master(full_table_name: str) -> str:
    """
    If the table name is that of a part table, then return what the master table name would be.
    This follows DataJoint's table naming convention where a master and a part must be in the
    same schema and the part table is prefixed with the master table name + ``__``.

    Example:
       `ephys`.`session`    -- master
       `ephys`.`session__recording`  -- part

    :param full_table_name: Full table name including part.
    :type full_table_name: str
    :return: Supposed master full table name or empty string if not a part table name.
    :rtype: str
    """
    match = re.match(r"(?P<master>`\w+`.`\w+)__(?P<part>\w+)`", full_table_name)
    return match["master"] + "`" if match else ""


def to_camel_case(s):
    """
    Convert names with under score (_) separation into camel case names.

    :param s: string in under_score notation
    :returns: string in CamelCase notation
    Example:
    >>> to_camel_case("table_name")  # returns "TableName"
    """

    def to_upper(match):
        return match.group(0)[-1].upper()

    return re.sub(r"(^|[_\W])+[a-zA-Z]", to_upper, s)


def from_camel_case(s):
    """
    Convert names in camel case into underscore (_) separated names

    :param s: string in CamelCase notation
    :returns: string in under_score notation
    Example:
    >>> from_camel_case("TableName") # yields "table_name"
    """

    def convert(match):
        return ("_" if match.groups()[0] else "") + match.group(0).lower()

    if not re.match(r"[A-Z][a-zA-Z0-9]*", s):
        raise DataJointError(
            "ClassName must be alphanumeric in CamelCase, begin with a capital letter"
        )
    return re.sub(r"(\B[A-Z])|(\b[A-Z])", convert, s)


def safe_write(filepath, blob):
    """
    A two-step write.

    :param filename: full path
    :param blob: binary data
    """
    filepath = Path(filepath)
    if not filepath.is_file():
        filepath.parent.mkdir(parents=True, exist_ok=True)
        temp_file = filepath.with_suffix(filepath.suffix + ".saving")
        temp_file.write_bytes(blob)
        temp_file.rename(filepath)


def safe_copy(src, dest, overwrite=False):
    """
    Copy the contents of src file into dest file as a two-step process. Skip if dest exists already
    """
    src, dest = Path(src), Path(dest)
    if not (dest.exists() and src.samefile(dest)) and (overwrite or not dest.is_file()):
        dest.parent.mkdir(parents=True, exist_ok=True)
        temp_file = dest.with_suffix(dest.suffix + ".copying")
        shutil.copyfile(str(src), str(temp_file))
        temp_file.rename(dest)


def parse_sql(filepath):
    """
    yield SQL statements from an SQL file
    """
    delimiter = ";"
    statement = []
    with Path(filepath).open("rt") as f:
        for line in f:
            line = line.strip()
            if not line.startswith("--") and len(line) > 1:
                if line.startswith("delimiter"):
                    delimiter = line.split()[1]
                else:
                    statement.append(line)
                    if line.endswith(delimiter):
                        yield " ".join(statement)
                        statement = []


def dict_to_uuid(key: dict):
    """Given a dictionary `key`, returns a hash string as UUID

    Args:
        key (dict): Any python dictionary"""
    hashed = hashlib.md5()
    for k, v in sorted(key.items()):
        hashed.update(str(k).encode())
        hashed.update(str(v).encode())
    return uuid.UUID(hex=hashed.hexdigest())

1	"""General-purpose utilities"""
2
3	import hashlib	1✔
4	import re	1✔
5	from pathlib import Path	1✔
6	import shutil	1✔
7	import uuid	1✔
8	from .errors import DataJointError	1✔
9	from .logging import logger	1✔
10
11
12	class ClassProperty:	1✔
13	def __init__(self, f):	1✔
14	self.f = f	1✔
15
16	def __get__(self, obj, owner):	1✔
17	return self.f(owner)	1✔
18
19
20	def user_choice(prompt, choices=("yes", "no"), default=None):	1✔
21	"""
22	Prompts the user for confirmation. The default value, if any, is capitalized.
23
24	:param prompt: Information to display to the user.
25	:param choices: an iterable of possible choices.
26	:param default: default choice
27	:return: the user's choice
28	"""
29	assert default is None or default in choices	1✔
30	choice_list = ", ".join(	1✔
31	(choice.title() if choice == default else choice for choice in choices)
32	)
33	response = None	1✔
34	while response not in choices:	1✔
35	response = input(prompt + " [" + choice_list + "]: ")	1✔
36	response = response.lower() if response else default	1✔
37	return response	1✔
38
39
40	def get_master(full_table_name: str) -> str:	1✔
41	"""
42	If the table name is that of a part table, then return what the master table name would be.
43	This follows DataJoint's table naming convention where a master and a part must be in the
44	same schema and the part table is prefixed with the master table name + ``__``.
45
46	Example:
47	`ephys`.`session` -- master
48	`ephys`.`session__recording` -- part
49
50	:param full_table_name: Full table name including part.
51	:type full_table_name: str
52	:return: Supposed master full table name or empty string if not a part table name.
53	:rtype: str
54	"""
55	match = re.match(r"(?P<master>`\w+`.`\w+)__(?P<part>\w+)`", full_table_name)	1✔
56	return match["master"] + "`" if match else ""	1✔
57
58
59	def to_camel_case(s):	1✔
60	"""
61	Convert names with under score (_) separation into camel case names.
62
63	:param s: string in under_score notation
64	:returns: string in CamelCase notation
65	Example:
66	>>> to_camel_case("table_name") # returns "TableName"
67	"""
68
69	def to_upper(match):	1✔
70	return match.group(0)[-1].upper()	1✔
71
72	return re.sub(r"(^\|[_\W])+[a-zA-Z]", to_upper, s)	1✔
73
74
75	def from_camel_case(s):	1✔
76	"""
77	Convert names in camel case into underscore (_) separated names
78
79	:param s: string in CamelCase notation
80	:returns: string in under_score notation
81	Example:
82	>>> from_camel_case("TableName") # yields "table_name"
83	"""
84
85	def convert(match):	1✔
86	return ("_" if match.groups()[0] else "") + match.group(0).lower()	1✔
87
88	if not re.match(r"[A-Z][a-zA-Z0-9]*", s):	1✔
89	raise DataJointError(	1✔
90	"ClassName must be alphanumeric in CamelCase, begin with a capital letter"
91	)
92	return re.sub(r"(\B[A-Z])\|(\b[A-Z])", convert, s)	1✔
93
94
95	def safe_write(filepath, blob):	1✔
96	"""
97	A two-step write.
98
99	:param filename: full path
100	:param blob: binary data
101	"""
102	filepath = Path(filepath)	1✔
103	if not filepath.is_file():	1✔
104	filepath.parent.mkdir(parents=True, exist_ok=True)	1✔
105	temp_file = filepath.with_suffix(filepath.suffix + ".saving")	1✔
106	temp_file.write_bytes(blob)	1✔
107	temp_file.rename(filepath)	1✔
108
109
110	def safe_copy(src, dest, overwrite=False):	1✔
111	"""
112	Copy the contents of src file into dest file as a two-step process. Skip if dest exists already
113	"""
114	src, dest = Path(src), Path(dest)	1✔
115	if not (dest.exists() and src.samefile(dest)) and (overwrite or not dest.is_file()):	1✔
116	dest.parent.mkdir(parents=True, exist_ok=True)	1✔
117	temp_file = dest.with_suffix(dest.suffix + ".copying")	1✔
118	shutil.copyfile(str(src), str(temp_file))	1✔
119	temp_file.rename(dest)	1✔
120
121
122	def parse_sql(filepath):	1✔
123	"""
124	yield SQL statements from an SQL file
125	"""
126	delimiter = ";"	1✔
127	statement = []	1✔
128	with Path(filepath).open("rt") as f:	1✔
129	for line in f:	1✔
130	line = line.strip()	1✔
131	if not line.startswith("--") and len(line) > 1:	1✔
132	if line.startswith("delimiter"):	1✔
133	delimiter = line.split()[1]	×
134	else:
135	statement.append(line)	1✔
136	if line.endswith(delimiter):	1✔
137	yield " ".join(statement)	1✔
138	statement = []	1✔
139
140
141	def dict_to_uuid(key: dict):	1✔
142	"""Given a dictionary `key`, returns a hash string as UUID
143
144	Args:
145	key (dict): Any python dictionary"""
146	hashed = hashlib.md5()	×
147	for k, v in sorted(key.items()):	×
148	hashed.update(str(k).encode())	×
149	hashed.update(str(v).encode())	×
150	return uuid.UUID(hex=hashed.hexdigest())	×

datajoint / datajoint-python / #12900

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