#12324

pending completion

Build # #12324

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/2913ac721646449e33ca11ab0cd85f336b116541">2913ac721<a href="https://github.com/datajoint/datajoint-python/commit/2913ac721646449e33ca11ab0cd85f336b116541">">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/4c1ee167f9aedbe96acec0ae445be504c04a35b3">4c1ee167f</a>">4c1ee167f</a><a href="https://github.com/datajoint/datajoint-python/commit/2913ac721646449e33ca11ab0cd85f336b116541"> into 0ff34f283">0ff34f283</a>

Run Details

9 of 9 new or added lines in 1 file covered. (100.0%)

3067 of 3392 relevant lines covered (90.42%)

0.9 hits per line

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

90.32

/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
import logging

logger = logging.getLogger(__name__)

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
        """
        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 (
                (
                    config.get("filepath_checksum_size_limit")  # check if None
                    and Path(local_filepath).stat().st_size
                    >= config.get("filepath_checksum_size_limit")
                )
                or uuid_from_file(local_filepath) == contents_hash
            )

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

datajoint / datajoint-python / #12324

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