#12362

pending completion

Build # #12362

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/dae17663737b33d71cea7182f829ab933fd460af">dae176637<a href="https://github.com/datajoint/datajoint-python/commit/dae17663737b33d71cea7182f829ab933fd460af">">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/761fab142e3ea1c960dafad2cbd21446a153f7d2">761fab142</a>">761fab142</a><a href="https://github.com/datajoint/datajoint-python/commit/dae17663737b33d71cea7182f829ab933fd460af"> into 0ff34f283">0ff34f283</a>

Run Details

11 of 11 new or added lines in 2 files covered. (100.0%)

2995 of 3320 relevant lines covered (90.21%)

0.9 hits per line

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

90.19

/datajoint/external.py

from pathlib import Path, PurePosixPath, PureWindowsPath
from collections.abc import Mapping
from tqdm import tqdm
from .settings import config
from .errors import DataJointError, MissingExternalFile
from .hash import uuid_from_buffer, uuid_from_file
from .table import Table, FreeTable
from .heading import Heading
from .declare import EXTERNAL_TABLE_ROOT
from . import s3
from .utils import safe_write, safe_copy

CACHE_SUBFOLDING = (
    2,
    2,
)  # (2, 2) means  "0123456789abcd" will be saved as "01/23/0123456789abcd"
SUPPORT_MIGRATED_BLOBS = True  # support blobs migrated from datajoint 0.11.*


def subfold(name, folds):
    """
    subfolding for external storage:   e.g.  subfold('aBCdefg', (2, 3))  -->  ['ab','cde']
    """
    return (
        (name[: folds[0]].lower(),) + subfold(name[folds[0] :], folds[1:])
        if folds
        else ()
    )


class ExternalTable(Table):
    """
    The table tracking externally stored objects.
    Declare as ExternalTable(connection, database)
    """

    def __init__(self, connection, store, database):
        self.store = store
        self.spec = config.get_store_spec(store)
        self._s3 = None
        self.database = database
        self._connection = connection
        self._heading = Heading(
            table_info=dict(
                conn=connection,
                database=database,
                table_name=self.table_name,
                context=None,
            )
        )
        self._support = [self.full_table_name]
        if not self.is_declared:
            self.declare()
        self._s3 = None
        if self.spec["protocol"] == "file" and not Path(self.spec["location"]).is_dir():
            raise FileNotFoundError(
                "Inaccessible local directory %s" % self.spec["location"]
            ) from None

    @property
    def definition(self):
        return """
        # external storage tracking
        hash  : uuid    #  hash of contents (blob), of filename + contents (attach), or relative filepath (filepath)
        ---
        size      :bigint unsigned     # size of object in bytes
        attachment_name=null : varchar(255)  # the filename of an attachment
        filepath=null : varchar(1000)  # relative filepath or attachment filename
        contents_hash=null : uuid      # used for the filepath datatype
        timestamp=CURRENT_TIMESTAMP  :timestamp   # automatic timestamp
        """

    @property
    def table_name(self):
        return "{external_table_root}_{store}".format(
            external_table_root=EXTERNAL_TABLE_ROOT, store=self.store
        )

    @property
    def s3(self):
        if self._s3 is None:
            self._s3 = s3.Folder(**self.spec)
        return self._s3

    # - low-level operations - private

    def _make_external_filepath(self, relative_filepath):
        """resolve the complete external path based on the relative path"""
        # Strip root
        if self.spec["protocol"] == "s3":
            posix_path = PurePosixPath(PureWindowsPath(self.spec["location"]))
            location_path = (
                Path(*posix_path.parts[1:])
                if len(self.spec["location"]) > 0
                and any(case in posix_path.parts[0] for case in ("\\", ":"))
                else Path(posix_path)
            )
            return PurePosixPath(location_path, relative_filepath)
        # Preserve root
        elif self.spec["protocol"] == "file":
            return PurePosixPath(Path(self.spec["location"]), relative_filepath)
        else:
            assert False

    def _make_uuid_path(self, uuid, suffix=""):
        """create external path based on the uuid hash"""
        return self._make_external_filepath(
            PurePosixPath(
                self.database,
                "/".join(subfold(uuid.hex, self.spec["subfolding"])),
                uuid.hex,
            ).with_suffix(suffix)
        )

    def _upload_file(self, local_path, external_path, metadata=None):
        if self.spec["protocol"] == "s3":
            self.s3.fput(local_path, external_path, metadata)
        elif self.spec["protocol"] == "file":
            safe_copy(local_path, external_path, overwrite=True)
        else:
            assert False

    def _download_file(self, external_path, download_path):
        if self.spec["protocol"] == "s3":
            self.s3.fget(external_path, download_path)
        elif self.spec["protocol"] == "file":
            safe_copy(external_path, download_path)
        else:
            assert False

    def _upload_buffer(self, buffer, external_path):
        if self.spec["protocol"] == "s3":
            self.s3.put(external_path, buffer)
        elif self.spec["protocol"] == "file":
            safe_write(external_path, buffer)
        else:
            assert False

    def _download_buffer(self, external_path):
        if self.spec["protocol"] == "s3":
            return self.s3.get(external_path)
        if self.spec["protocol"] == "file":
            return Path(external_path).read_bytes()
        assert False

    def _remove_external_file(self, external_path):
        if self.spec["protocol"] == "s3":
            self.s3.remove_object(external_path)
        elif self.spec["protocol"] == "file":
            try:
                Path(external_path).unlink()
            except FileNotFoundError:
                pass

    def exists(self, external_filepath):
        """
        :return: True if the external file is accessible
        """
        if self.spec["protocol"] == "s3":
            return self.s3.exists(external_filepath)
        if self.spec["protocol"] == "file":
            return Path(external_filepath).is_file()
        assert False

    # --- BLOBS ----

    def put(self, blob):
        """
        put a binary string (blob) in external store
        """
        uuid = uuid_from_buffer(blob)
        self._upload_buffer(blob, self._make_uuid_path(uuid))
        # insert tracking info
        self.connection.query(
            "INSERT INTO {tab} (hash, size) VALUES (%s, {size}) ON DUPLICATE KEY "
            "UPDATE timestamp=CURRENT_TIMESTAMP".format(
                tab=self.full_table_name, size=len(blob)
            ),
            args=(uuid.bytes,),
        )
        return uuid

    def get(self, uuid):
        """
        get an object from external store.
        """
        if uuid is None:
            return None
        # attempt to get object from cache
        blob = None
        cache_folder = config.get("cache", None)
        if cache_folder:
            try:
                cache_path = Path(cache_folder, *subfold(uuid.hex, CACHE_SUBFOLDING))
                cache_file = Path(cache_path, uuid.hex)
                blob = cache_file.read_bytes()
            except FileNotFoundError:
                pass  # not cached
        # download blob from external store
        if blob is None:
            try:
                blob = self._download_buffer(self._make_uuid_path(uuid))
            except MissingExternalFile:
                if not SUPPORT_MIGRATED_BLOBS:
                    raise
                # blobs migrated from datajoint 0.11 are stored at explicitly defined filepaths
                relative_filepath, contents_hash = (self & {"hash": uuid}).fetch1(
                    "filepath", "contents_hash"
                )
                if relative_filepath is None:
                    raise
                blob = self._download_buffer(
                    self._make_external_filepath(relative_filepath)
                )
            if cache_folder:
                cache_path.mkdir(parents=True, exist_ok=True)
                safe_write(cache_path / uuid.hex, blob)
        return blob

    # --- ATTACHMENTS ---

    def upload_attachment(self, local_path):
        attachment_name = Path(local_path).name
        uuid = uuid_from_file(local_path, init_string=attachment_name + "\0")
        external_path = self._make_uuid_path(uuid, "." + attachment_name)
        self._upload_file(local_path, external_path)
        # insert tracking info
        self.connection.query(
            """
        INSERT INTO {tab} (hash, size, attachment_name)
        VALUES (%s, {size}, "{attachment_name}")
        ON DUPLICATE KEY UPDATE timestamp=CURRENT_TIMESTAMP""".format(
                tab=self.full_table_name,
                size=Path(local_path).stat().st_size,
                attachment_name=attachment_name,
            ),
            args=[uuid.bytes],
        )
        return uuid

    def get_attachment_name(self, uuid):
        return (self & {"hash": uuid}).fetch1("attachment_name")

    def download_attachment(self, uuid, attachment_name, download_path):
        """save attachment from memory buffer into the save_path"""
        external_path = self._make_uuid_path(uuid, "." + attachment_name)
        self._download_file(external_path, download_path)

    # --- FILEPATH ---

    def upload_filepath(self, local_filepath):
        """
        Raise exception if an external entry already exists with a different contents checksum.
        Otherwise, copy (with overwrite) file to remote and
        If an external entry exists with the same checksum, then no copying should occur
        """
        local_filepath = Path(local_filepath)
        try:
            relative_filepath = str(
                local_filepath.relative_to(self.spec["stage"]).as_posix()
            )
        except ValueError:
            raise DataJointError(
                "The path {path} is not in stage {stage}".format(
                    path=local_filepath.parent, **self.spec
                )
            )
        uuid = uuid_from_buffer(
            init_string=relative_filepath
        )  # hash relative path, not contents
        contents_hash = uuid_from_file(local_filepath)

        # check if the remote file already exists and verify that it matches
        check_hash = (self & {"hash": uuid}).fetch("contents_hash")
        if check_hash:
            # the tracking entry exists, check that it's the same file as before
            if contents_hash != check_hash[0]:
                raise DataJointError(
                    "A different version of '{file}' has already been placed.".format(
                        file=relative_filepath
                    )
                )
        else:
            # upload the file and create its tracking entry
            self._upload_file(
                local_filepath,
                self._make_external_filepath(relative_filepath),
                metadata={"contents_hash": str(contents_hash)},
            )
            self.connection.query(
                "INSERT INTO {tab} (hash, size, filepath, contents_hash) VALUES (%s, {size}, '{filepath}', %s)".format(
                    tab=self.full_table_name,
                    size=Path(local_filepath).stat().st_size,
                    filepath=relative_filepath,
                ),
                args=(uuid.bytes, contents_hash.bytes),
            )
        return uuid

    def download_filepath(self, filepath_hash):
        """
        sync a file from external store to the local stage

        :param filepath_hash: The hash (UUID) of the relative_path
        :return: hash (UUID) of the contents of the downloaded file or Nones
        """

        def _need_checksum(local_filepath):
            limit = config.get("filepath_checksum_size_limit")
            return limit is None or Path(local_filepath).stat().st_size < limit

        if filepath_hash is not None:
            relative_filepath, contents_hash = (self & {"hash": filepath_hash}).fetch1(
                "filepath", "contents_hash"
            )
            external_path = self._make_external_filepath(relative_filepath)
            local_filepath = Path(self.spec["stage"]).absolute() / relative_filepath

            file_exists = Path(local_filepath).is_file() and (
                not _need_checksum(local_filepath)
                or uuid_from_file(local_filepath) == contents_hash
            )

            if not file_exists:
                self._download_file(external_path, local_filepath)
                if _need_checksum(local_filepath):
                    checksum = uuid_from_file(local_filepath)
                    if checksum != contents_hash:
                        # this should never happen without outside interference
                        raise DataJointError(
                            "'{file}' downloaded but did not pass checksum'".format(
                                file=local_filepath
                            )
                        )
            if not _need_checksum(local_filepath):
                print(f"WARNING SKIPPED CHECKSUM FOR FILE WITH HASH: {contents_hash}")
                # This will turn into a proper logger when we implement the datajoint logger
            return str(local_filepath), contents_hash

    # --- UTILITIES ---

    @property
    def references(self):
        """
        :return: generator of referencing table names and their referencing columns
        """
        return (
            {k.lower(): v for k, v in elem.items()}
            for elem in self.connection.query(
                """
        SELECT concat('`', table_schema, '`.`', table_name, '`') as referencing_table, column_name
        FROM information_schema.key_column_usage
        WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
        """.format(
                    tab=self.table_name, db=self.database
                ),
                as_dict=True,
            )
        )

    def fetch_external_paths(self, **fetch_kwargs):
        """
        generate complete external filepaths from the query.
        Each element is a tuple: (uuid, path)

        :param fetch_kwargs: keyword arguments to pass to fetch
        """
        fetch_kwargs.update(as_dict=True)
        paths = []
        for item in self.fetch("hash", "attachment_name", "filepath", **fetch_kwargs):
            if item["attachment_name"]:
                # attachments
                path = self._make_uuid_path(item["hash"], "." + item["attachment_name"])
            elif item["filepath"]:
                # external filepaths
                path = self._make_external_filepath(item["filepath"])
            else:
                # blobs
                path = self._make_uuid_path(item["hash"])
            paths.append((item["hash"], path))
        return paths

    def unused(self):
        """
        query expression for unused hashes

        :return: self restricted to elements that are not in use by any tables in the schema
        """
        return self - [
            FreeTable(self.connection, ref["referencing_table"]).proj(
                hash=ref["column_name"]
            )
            for ref in self.references
        ]

    def used(self):
        """
        query expression for used hashes

        :return: self restricted to elements that in use by tables in the schema
        """
        return self & [
            FreeTable(self.connection, ref["referencing_table"]).proj(
                hash=ref["column_name"]
            )
            for ref in self.references
        ]

    def delete(
        self,
        *,
        delete_external_files=None,
        limit=None,
        display_progress=True,
        errors_as_string=True,
    ):
        """

        :param delete_external_files: True or False. If False, only the tracking info is removed from the external
                store table but the external files remain intact. If True, then the external files themselves are deleted too.
        :param errors_as_string: If True any errors returned when deleting from external files will be strings
        :param limit: (integer) limit the number of items to delete
        :param display_progress: if True, display progress as files are cleaned up
        :return: if deleting external files, returns errors
        """
        if delete_external_files not in (True, False):
            raise DataJointError(
                "The delete_external_files argument must be set to either "
                "True or False in delete()"
            )

        if not delete_external_files:
            self.unused().delete_quick()
        else:
            items = self.unused().fetch_external_paths(limit=limit)
            if display_progress:
                items = tqdm(items)
            # delete items one by one, close to transaction-safe
            error_list = []
            for uuid, external_path in items:
                row = (self & {"hash": uuid}).fetch()
                if row.size:
                    try:
                        (self & {"hash": uuid}).delete_quick()
                    except Exception:
                        pass  # if delete failed, do not remove the external file
                    else:
                        try:
                            self._remove_external_file(external_path)
                        except Exception as error:
                            # adding row back into table after failed delete
                            self.insert1(row[0], skip_duplicates=True)
                            error_list.append(
                                (
                                    uuid,
                                    external_path,
                                    str(error) if errors_as_string else error,
                                )
                            )
            return error_list


class ExternalMapping(Mapping):
    """
    The external manager contains all the tables for all external stores for a given schema
    :Example:
        e = ExternalMapping(schema)
        external_table = e[store]
    """

    def __init__(self, schema):
        self.schema = schema
        self._tables = {}

    def __repr__(self):
        return "External file tables for schema `{schema}`:\n    ".format(
            schema=self.schema.database
        ) + "\n    ".join(
            '"{store}" {protocol}:{location}'.format(store=k, **v.spec)
            for k, v in self.items()
        )

    def __getitem__(self, store):
        """
        Triggers the creation of an external table.
        Should only be used when ready to save or read from external storage.

        :param store: the name of the store
        :return: the ExternalTable object for the store
        """
        if store not in self._tables:
            self._tables[store] = ExternalTable(
                connection=self.schema.connection,
                store=store,
                database=self.schema.database,
            )
        return self._tables[store]

    def __len__(self):
        return len(self._tables)

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

1	from pathlib import Path, PurePosixPath, PureWindowsPath	1✔
2	from collections.abc import Mapping	1✔
3	from tqdm import tqdm	1✔
4	from .settings import config	1✔
5	from .errors import DataJointError, MissingExternalFile	1✔
6	from .hash import uuid_from_buffer, uuid_from_file	1✔
7	from .table import Table, FreeTable	1✔
8	from .heading import Heading	1✔
9	from .declare import EXTERNAL_TABLE_ROOT	1✔
10	from . import s3	1✔
11	from .utils import safe_write, safe_copy	1✔
12
13	CACHE_SUBFOLDING = (	1✔
14	2,
15	2,
16	) # (2, 2) means "0123456789abcd" will be saved as "01/23/0123456789abcd"
17	SUPPORT_MIGRATED_BLOBS = True # support blobs migrated from datajoint 0.11.*	1✔
18
19
20	def subfold(name, folds):	1✔
21	"""
22	subfolding for external storage: e.g. subfold('aBCdefg', (2, 3)) --> ['ab','cde']
23	"""
24	return (	1✔
25	(name[: folds[0]].lower(),) + subfold(name[folds[0] :], folds[1:])
26	if folds
27	else ()
28	)
29
30
31	class ExternalTable(Table):	1✔
32	"""
33	The table tracking externally stored objects.
34	Declare as ExternalTable(connection, database)
35	"""
36
37	def __init__(self, connection, store, database):	1✔
38	self.store = store	1✔
39	self.spec = config.get_store_spec(store)	1✔
40	self._s3 = None	1✔
41	self.database = database	1✔
42	self._connection = connection	1✔
43	self._heading = Heading(	1✔
44	table_info=dict(
45	conn=connection,
46	database=database,
47	table_name=self.table_name,
48	context=None,
49	)
50	)
51	self._support = [self.full_table_name]	1✔
52	if not self.is_declared:	1✔
53	self.declare()	1✔
54	self._s3 = None	1✔
55	if self.spec["protocol"] == "file" and not Path(self.spec["location"]).is_dir():	1✔
56	raise FileNotFoundError(	×
57	"Inaccessible local directory %s" % self.spec["location"]
58	) from None
59
60	@property	1✔
61	def definition(self):
62	return """	1✔
63	# external storage tracking
64	hash : uuid # hash of contents (blob), of filename + contents (attach), or relative filepath (filepath)
65	---
66	size :bigint unsigned # size of object in bytes
67	attachment_name=null : varchar(255) # the filename of an attachment
68	filepath=null : varchar(1000) # relative filepath or attachment filename
69	contents_hash=null : uuid # used for the filepath datatype
70	timestamp=CURRENT_TIMESTAMP :timestamp # automatic timestamp
71	"""
72
73	@property	1✔
74	def table_name(self):
75	return "{external_table_root}_{store}".format(	1✔
76	external_table_root=EXTERNAL_TABLE_ROOT, store=self.store
77	)
78
79	@property	1✔
80	def s3(self):
81	if self._s3 is None:	1✔
82	self._s3 = s3.Folder(**self.spec)	1✔
83	return self._s3	1✔
84
85	# - low-level operations - private
86
87	def _make_external_filepath(self, relative_filepath):	1✔
88	"""resolve the complete external path based on the relative path"""
89	# Strip root
90	if self.spec["protocol"] == "s3":	1✔
91	posix_path = PurePosixPath(PureWindowsPath(self.spec["location"]))	1✔
92	location_path = (	1✔
93	Path(*posix_path.parts[1:])
94	if len(self.spec["location"]) > 0
95	and any(case in posix_path.parts[0] for case in ("\\", ":"))
96	else Path(posix_path)
97	)
98	return PurePosixPath(location_path, relative_filepath)	1✔
99	# Preserve root
100	elif self.spec["protocol"] == "file":	1✔
101	return PurePosixPath(Path(self.spec["location"]), relative_filepath)	1✔
102	else:
103	assert False	×
104
105	def _make_uuid_path(self, uuid, suffix=""):	1✔
106	"""create external path based on the uuid hash"""
107	return self._make_external_filepath(	1✔
108	PurePosixPath(
109	self.database,
110	"/".join(subfold(uuid.hex, self.spec["subfolding"])),
111	uuid.hex,
112	).with_suffix(suffix)
113	)
114
115	def _upload_file(self, local_path, external_path, metadata=None):	1✔
116	if self.spec["protocol"] == "s3":	1✔
117	self.s3.fput(local_path, external_path, metadata)	1✔
118	elif self.spec["protocol"] == "file":	1✔
119	safe_copy(local_path, external_path, overwrite=True)	1✔
120	else:
121	assert False	×
122
123	def _download_file(self, external_path, download_path):	1✔
124	if self.spec["protocol"] == "s3":	1✔
125	self.s3.fget(external_path, download_path)	1✔
126	elif self.spec["protocol"] == "file":	1✔
127	safe_copy(external_path, download_path)	1✔
128	else:
129	assert False	×
130
131	def _upload_buffer(self, buffer, external_path):	1✔
132	if self.spec["protocol"] == "s3":	1✔
133	self.s3.put(external_path, buffer)	1✔
134	elif self.spec["protocol"] == "file":	1✔
135	safe_write(external_path, buffer)	1✔
136	else:
137	assert False	×
138
139	def _download_buffer(self, external_path):	1✔
140	if self.spec["protocol"] == "s3":	1✔
141	return self.s3.get(external_path)	1✔
142	if self.spec["protocol"] == "file":	1✔
143	return Path(external_path).read_bytes()	1✔
144	assert False	×
145
146	def _remove_external_file(self, external_path):	1✔
147	if self.spec["protocol"] == "s3":	1✔
148	self.s3.remove_object(external_path)	1✔
149	elif self.spec["protocol"] == "file":	1✔
150	try:	1✔
151	Path(external_path).unlink()	1✔
152	except FileNotFoundError:	1✔
153	pass	×
154
155	def exists(self, external_filepath):	1✔
156	"""
157	:return: True if the external file is accessible
158	"""
159	if self.spec["protocol"] == "s3":	1✔
160	return self.s3.exists(external_filepath)	1✔
161	if self.spec["protocol"] == "file":	1✔
162	return Path(external_filepath).is_file()	1✔
163	assert False	×
164
165	# --- BLOBS ----
166
167	def put(self, blob):	1✔
168	"""
169	put a binary string (blob) in external store
170	"""
171	uuid = uuid_from_buffer(blob)	1✔
172	self._upload_buffer(blob, self._make_uuid_path(uuid))	1✔
173	# insert tracking info
174	self.connection.query(	1✔
175	"INSERT INTO {tab} (hash, size) VALUES (%s, {size}) ON DUPLICATE KEY "
176	"UPDATE timestamp=CURRENT_TIMESTAMP".format(
177	tab=self.full_table_name, size=len(blob)
178	),
179	args=(uuid.bytes,),
180	)
181	return uuid	1✔
182
183	def get(self, uuid):	1✔
184	"""
185	get an object from external store.
186	"""
187	if uuid is None:	1✔
188	return None	×
189	# attempt to get object from cache
190	blob = None	1✔
191	cache_folder = config.get("cache", None)	1✔
192	if cache_folder:	1✔
193	try:	1✔
194	cache_path = Path(cache_folder, *subfold(uuid.hex, CACHE_SUBFOLDING))	1✔
195	cache_file = Path(cache_path, uuid.hex)	1✔
196	blob = cache_file.read_bytes()	1✔
197	except FileNotFoundError:	1✔
198	pass # not cached	1✔
199	# download blob from external store
200	if blob is None:	1✔
201	try:	1✔
202	blob = self._download_buffer(self._make_uuid_path(uuid))	1✔
203	except MissingExternalFile:	1✔
204	if not SUPPORT_MIGRATED_BLOBS:	1✔
205	raise	×
206	# blobs migrated from datajoint 0.11 are stored at explicitly defined filepaths
207	relative_filepath, contents_hash = (self & {"hash": uuid}).fetch1(	1✔
208	"filepath", "contents_hash"
209	)
210	if relative_filepath is None:	1✔
211	raise	×
212	blob = self._download_buffer(	1✔
213	self._make_external_filepath(relative_filepath)
214	)
215	if cache_folder:	1✔
216	cache_path.mkdir(parents=True, exist_ok=True)	1✔
217	safe_write(cache_path / uuid.hex, blob)	1✔
218	return blob	1✔
219
220	# --- ATTACHMENTS ---
221
222	def upload_attachment(self, local_path):	1✔
223	attachment_name = Path(local_path).name	1✔
224	uuid = uuid_from_file(local_path, init_string=attachment_name + "\0")	1✔
225	external_path = self._make_uuid_path(uuid, "." + attachment_name)	1✔
226	self._upload_file(local_path, external_path)	1✔
227	# insert tracking info
228	self.connection.query(	1✔
229	"""
230	INSERT INTO {tab} (hash, size, attachment_name)
231	VALUES (%s, {size}, "{attachment_name}")
232	ON DUPLICATE KEY UPDATE timestamp=CURRENT_TIMESTAMP""".format(
233	tab=self.full_table_name,
234	size=Path(local_path).stat().st_size,
235	attachment_name=attachment_name,
236	),
237	args=[uuid.bytes],
238	)
239	return uuid	1✔
240
241	def get_attachment_name(self, uuid):	1✔
242	return (self & {"hash": uuid}).fetch1("attachment_name")	1✔
243
244	def download_attachment(self, uuid, attachment_name, download_path):	1✔
245	"""save attachment from memory buffer into the save_path"""
246	external_path = self._make_uuid_path(uuid, "." + attachment_name)	1✔
247	self._download_file(external_path, download_path)	1✔
248
249	# --- FILEPATH ---
250
251	def upload_filepath(self, local_filepath):	1✔
252	"""
253	Raise exception if an external entry already exists with a different contents checksum.
254	Otherwise, copy (with overwrite) file to remote and
255	If an external entry exists with the same checksum, then no copying should occur
256	"""
257	local_filepath = Path(local_filepath)	1✔
258	try:	1✔
259	relative_filepath = str(	1✔
260	local_filepath.relative_to(self.spec["stage"]).as_posix()
261	)
262	except ValueError:	×
263	raise DataJointError(	×
264	"The path {path} is not in stage {stage}".format(
265	path=local_filepath.parent, **self.spec
266	)
267	)
268	uuid = uuid_from_buffer(	1✔
269	init_string=relative_filepath
270	) # hash relative path, not contents
271	contents_hash = uuid_from_file(local_filepath)	1✔
272
273	# check if the remote file already exists and verify that it matches
274	check_hash = (self & {"hash": uuid}).fetch("contents_hash")	1✔
275	if check_hash:	1✔
276	# the tracking entry exists, check that it's the same file as before
277	if contents_hash != check_hash[0]:	1✔
278	raise DataJointError(	1✔
279	"A different version of '{file}' has already been placed.".format(
280	file=relative_filepath
281	)
282	)
283	else:
284	# upload the file and create its tracking entry
285	self._upload_file(	1✔
286	local_filepath,
287	self._make_external_filepath(relative_filepath),
288	metadata={"contents_hash": str(contents_hash)},
289	)
290	self.connection.query(	1✔
291	"INSERT INTO {tab} (hash, size, filepath, contents_hash) VALUES (%s, {size}, '{filepath}', %s)".format(
292	tab=self.full_table_name,
293	size=Path(local_filepath).stat().st_size,
294	filepath=relative_filepath,
295	),
296	args=(uuid.bytes, contents_hash.bytes),
297	)
298	return uuid	1✔
299
300	def download_filepath(self, filepath_hash):	1✔
301	"""
302	sync a file from external store to the local stage
303
304	:param filepath_hash: The hash (UUID) of the relative_path
305	:return: hash (UUID) of the contents of the downloaded file or Nones
306	"""
307
308	def _need_checksum(local_filepath):	1✔
309	limit = config.get("filepath_checksum_size_limit")	1✔
310	return limit is None or Path(local_filepath).stat().st_size < limit	1✔
311
312	if filepath_hash is not None:	1✔
313	relative_filepath, contents_hash = (self & {"hash": filepath_hash}).fetch1(	1✔
314	"filepath", "contents_hash"
315	)
316	external_path = self._make_external_filepath(relative_filepath)	1✔
317	local_filepath = Path(self.spec["stage"]).absolute() / relative_filepath	1✔
318
319	file_exists = Path(local_filepath).is_file() and (	1✔
320	not _need_checksum(local_filepath)
321	or uuid_from_file(local_filepath) == contents_hash
322	)
323
324	if not file_exists:	1✔
325	self._download_file(external_path, local_filepath)	1✔
326	if _need_checksum(local_filepath):	1✔
327	checksum = uuid_from_file(local_filepath)	1✔
328	if checksum != contents_hash:	1✔
329	# this should never happen without outside interference
330	raise DataJointError(	×
331	"'{file}' downloaded but did not pass checksum'".format(
332	file=local_filepath
333	)
334	)
335	if not _need_checksum(local_filepath):	1✔
336	print(f"WARNING SKIPPED CHECKSUM FOR FILE WITH HASH: {contents_hash}")	1✔
337	# This will turn into a proper logger when we implement the datajoint logger
338	return str(local_filepath), contents_hash	1✔
339
340	# --- UTILITIES ---
341
342	@property	1✔
343	def references(self):
344	"""
345	:return: generator of referencing table names and their referencing columns
346	"""
347	return (	1✔
348	{k.lower(): v for k, v in elem.items()}
349	for elem in self.connection.query(
350	"""
351	SELECT concat('`', table_schema, '`.`', table_name, '`') as referencing_table, column_name
352	FROM information_schema.key_column_usage
353	WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
354	""".format(
355	tab=self.table_name, db=self.database
356	),
357	as_dict=True,
358	)
359	)
360
361	def fetch_external_paths(self, **fetch_kwargs):	1✔
362	"""
363	generate complete external filepaths from the query.
364	Each element is a tuple: (uuid, path)
365
366	:param fetch_kwargs: keyword arguments to pass to fetch
367	"""
368	fetch_kwargs.update(as_dict=True)	1✔
369	paths = []	1✔
370	for item in self.fetch("hash", "attachment_name", "filepath", **fetch_kwargs):	1✔
371	if item["attachment_name"]:	1✔
372	# attachments
373	path = self._make_uuid_path(item["hash"], "." + item["attachment_name"])	×
374	elif item["filepath"]:	1✔
375	# external filepaths
376	path = self._make_external_filepath(item["filepath"])	1✔
377	else:
378	# blobs
379	path = self._make_uuid_path(item["hash"])	1✔
380	paths.append((item["hash"], path))	1✔
381	return paths	1✔
382
383	def unused(self):	1✔
384	"""
385	query expression for unused hashes
386
387	:return: self restricted to elements that are not in use by any tables in the schema
388	"""
389	return self - [	1✔
390	FreeTable(self.connection, ref["referencing_table"]).proj(
391	hash=ref["column_name"]
392	)
393	for ref in self.references
394	]
395
396	def used(self):	1✔
397	"""
398	query expression for used hashes
399
400	:return: self restricted to elements that in use by tables in the schema
401	"""
402	return self & [	×
403	FreeTable(self.connection, ref["referencing_table"]).proj(
404	hash=ref["column_name"]
405	)
406	for ref in self.references
407	]
408
409	def delete(	1✔
410	self,
411	*,
412	delete_external_files=None,
413	limit=None,
414	display_progress=True,
415	errors_as_string=True,
416	):
417	"""
418
419	:param delete_external_files: True or False. If False, only the tracking info is removed from the external
420	store table but the external files remain intact. If True, then the external files themselves are deleted too.
421	:param errors_as_string: If True any errors returned when deleting from external files will be strings
422	:param limit: (integer) limit the number of items to delete
423	:param display_progress: if True, display progress as files are cleaned up
424	:return: if deleting external files, returns errors
425	"""
426	if delete_external_files not in (True, False):	1✔
427	raise DataJointError(	×
428	"The delete_external_files argument must be set to either "
429	"True or False in delete()"
430	)
431
432	if not delete_external_files:	1✔
433	self.unused().delete_quick()	1✔
434	else:
435	items = self.unused().fetch_external_paths(limit=limit)	1✔
436	if display_progress:	1✔
437	items = tqdm(items)	1✔
438	# delete items one by one, close to transaction-safe
439	error_list = []	1✔
440	for uuid, external_path in items:	1✔
441	row = (self & {"hash": uuid}).fetch()	1✔
442	if row.size:	1✔
443	try:	1✔
444	(self & {"hash": uuid}).delete_quick()	1✔
445	except Exception:	×
446	pass # if delete failed, do not remove the external file	×
447	else:
448	try:	1✔
449	self._remove_external_file(external_path)	1✔
450	except Exception as error:	1✔
451	# adding row back into table after failed delete
452	self.insert1(row[0], skip_duplicates=True)	1✔
453	error_list.append(	1✔
454	(
455	uuid,
456	external_path,
457	str(error) if errors_as_string else error,
458	)
459	)
460	return error_list	1✔
461
462
463	class ExternalMapping(Mapping):	1✔
464	"""
465	The external manager contains all the tables for all external stores for a given schema
466	:Example:
467	e = ExternalMapping(schema)
468	external_table = e[store]
469	"""
470
471	def __init__(self, schema):	1✔
472	self.schema = schema	1✔
473	self._tables = {}	1✔
474
475	def __repr__(self):	1✔
476	return "External file tables for schema `{schema}`:\n ".format(	×
477	schema=self.schema.database
478	) + "\n ".join(
479	'"{store}" {protocol}:{location}'.format(store=k, **v.spec)
480	for k, v in self.items()
481	)
482
483	def __getitem__(self, store):	1✔
484	"""
485	Triggers the creation of an external table.
486	Should only be used when ready to save or read from external storage.
487
488	:param store: the name of the store
489	:return: the ExternalTable object for the store
490	"""
491	if store not in self._tables:	1✔
492	self._tables[store] = ExternalTable(	1✔
493	connection=self.schema.connection,
494	store=store,
495	database=self.schema.database,
496	)
497	return self._tables[store]	1✔
498
499	def __len__(self):	1✔
500	return len(self._tables)	×
501
502	def __iter__(self):	1✔
503	return iter(self._tables)	1✔

datajoint / datajoint-python / #12362

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