#12010

pending completion

Build # #12010

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/1fda6c8a1714736a5139efa6dc2e572836c19398">1fda6c8a1<a href="https://github.com/datajoint/datajoint-python/commit/1fda6c8a1714736a5139efa6dc2e572836c19398">">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/8c51bde61f106c71254a2cb6d0c9ed7b22acc9fe">8c51bde61</a>">8c51bde61</a><a href="https://github.com/datajoint/datajoint-python/commit/1fda6c8a1714736a5139efa6dc2e572836c19398"> into f5fc2f68d">f5fc2f68d</a>

Run Details

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

98.25

/datajoint/utils.py

"""General-purpose utilities"""

import re
from pathlib import Path
import shutil
from .errors import DataJointError


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 = []

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

datajoint / datajoint-python / #12010

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