#12425

pending completion

Build # #12425

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/2de6025ca1a1505e8b55128e3bf7a211ef5ae0e4">2de6025ca<a href="https://github.com/datajoint/datajoint-python/commit/2de6025ca1a1505e8b55128e3bf7a211ef5ae0e4">">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/1ecbe5fdc3d4295c5ac1be432c0739ec53ec771d">1ecbe5fdc</a>">1ecbe5fdc</a><a href="https://github.com/datajoint/datajoint-python/commit/2de6025ca1a1505e8b55128e3bf7a211ef5ae0e4"> into 0ff34f283">0ff34f283</a>

Run Details

77 of 77 new or added lines in 16 files covered. (100.0%)

3093 of 3430 relevant lines covered (90.17%)

0.9 hits per line

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

90.05

/datajoint/external.py

from pathlib import Path, PurePosixPath, PureWindowsPath
from collections.abc import Mapping
from tqdm import tqdm
import logging
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

logger = logging.getLogger(__name__.split(".")[0])

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

datajoint / datajoint-python / #12425

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