#12391

pending completion

Build # #12391

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/6fd979380c2ca4b0c46e26f68bc1963a4b0271f0">6fd979380<a href="https://github.com/datajoint/datajoint-python/commit/6fd979380c2ca4b0c46e26f68bc1963a4b0271f0">">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/053c0fc4d5047faa3ecbbe365215c185452af6e1">053c0fc4d</a>">053c0fc4d</a><a href="https://github.com/datajoint/datajoint-python/commit/6fd979380c2ca4b0c46e26f68bc1963a4b0271f0"> into 0ff34f283">0ff34f283</a>

Run Details

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

3068 of 3393 relevant lines covered (90.42%)

0.9 hits per line

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

90.37

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

datajoint / datajoint-python / #12391

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