#12897

pending completion

Build # #12897

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/715ab40552f63cd79723ed2830c6691b2cb228b9">715ab4055<a href="https://github.com/datajoint/datajoint-python/commit/715ab40552f63cd79723ed2830c6691b2cb228b9">">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/0a4f193031d8b1e14b09ec62d83c5def3b7421b0">0a4f19303</a>">0a4f19303</a><a href="https://github.com/datajoint/datajoint-python/commit/715ab40552f63cd79723ed2830c6691b2cb228b9"> into 3b6e84588">3b6e84588</a>

Run Details

69 of 69 new or added lines in 9 files covered. (100.0%)

3125 of 3454 relevant lines covered (90.47%)

0.9 hits per line

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

82.0

/datajoint/migrate.py

import datajoint as dj
from pathlib import Path
import re
from .utils import user_choice
import logging

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


def migrate_dj011_external_blob_storage_to_dj012(migration_schema, store):
    """
    Utility function to migrate external blob data from 0.11 to 0.12.

    :param migration_schema: string of target schema to be migrated
    :param store: string of target dj.config['store'] to be migrated
    """
    if not isinstance(migration_schema, str):
        raise ValueError(
            "Expected type {} for migration_schema, not {}.".format(
                str, type(migration_schema)
            )
        )

    do_migration = False
    do_migration = (
        user_choice(
            """
Warning: Ensure the following are completed before proceeding.
- Appropriate backups have been taken,
- Any existing DJ 0.11.X connections are suspended, and
- External config has been updated to new dj.config['stores'] structure.
Proceed?
            """,
            default="no",
        )
        == "yes"
    )
    if do_migration:
        _migrate_dj011_blob(dj.Schema(migration_schema), store)
        logger.info(
            "Migration completed for schema: {}, store: {}.".format(
                migration_schema, store
            )
        )
        return
    logger.warn("No migration performed.")


def _migrate_dj011_blob(schema, default_store):
    query = schema.connection.query

    LEGACY_HASH_SIZE = 43

    legacy_external = dj.FreeTable(
        schema.connection, "`{db}`.`~external`".format(db=schema.database)
    )

    # get referencing tables
    refs = [
        {k.lower(): v for k, v in elem.items()}
        for elem in query(
            """
    SELECT concat('`', table_schema, '`.`', table_name, '`')
            as referencing_table, column_name, constraint_name
    FROM information_schema.key_column_usage
    WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
    """.format(
                tab=legacy_external.table_name, db=legacy_external.database
            ),
            as_dict=True,
        ).fetchall()
    ]

    for ref in refs:
        # get comment
        column = query(
            "SHOW FULL COLUMNS FROM {referencing_table}"
            'WHERE Field="{column_name}"'.format(**ref),
            as_dict=True,
        ).fetchone()

        store, comment = re.match(
            r":external(-(?P<store>.+))?:(?P<comment>.*)", column["Comment"]
        ).group("store", "comment")

        # get all the hashes from the reference
        hashes = {
            x[0]
            for x in query(
                "SELECT `{column_name}` FROM {referencing_table}".format(**ref)
            )
        }

        # sanity check make sure that store suffixes match
        if store is None:
            assert all(len(_) == LEGACY_HASH_SIZE for _ in hashes)
        else:
            assert all(_[LEGACY_HASH_SIZE:] == store for _ in hashes)

        # create new-style external table
        ext = schema.external[store or default_store]

        # add the new-style reference field
        temp_suffix = "tempsub"

        try:
            query(
                """ALTER TABLE {referencing_table}
                ADD COLUMN `{column_name}_{temp_suffix}` {type} DEFAULT NULL
            COMMENT ":blob@{store}:{comment}"
            """.format(
                    type=dj.declare.UUID_DATA_TYPE,
                    temp_suffix=temp_suffix,
                    store=(store or default_store),
                    comment=comment,
                    **ref
                )
            )
        except:
            logger.warn("Column already added")

        for _hash, size in zip(*legacy_external.fetch("hash", "size")):
            if _hash in hashes:
                relative_path = str(Path(schema.database, _hash).as_posix())
                uuid = dj.hash.uuid_from_buffer(init_string=relative_path)
                external_path = ext._make_external_filepath(relative_path)
                if ext.spec["protocol"] == "s3":
                    contents_hash = dj.hash.uuid_from_buffer(
                        ext._download_buffer(external_path)
                    )
                else:
                    contents_hash = dj.hash.uuid_from_file(external_path)
                ext.insert1(
                    dict(
                        filepath=relative_path,
                        size=size,
                        contents_hash=contents_hash,
                        hash=uuid,
                    ),
                    skip_duplicates=True,
                )

                query(
                    "UPDATE {referencing_table} "
                    "SET `{column_name}_{temp_suffix}`=%s "
                    'WHERE `{column_name}` = "{_hash}"'.format(
                        _hash=_hash, temp_suffix=temp_suffix, **ref
                    ),
                    uuid.bytes,
                )

        # check that all have been copied
        check = query(
            "SELECT * FROM {referencing_table} "
            "WHERE `{column_name}` IS NOT NULL"
            "  AND `{column_name}_{temp_suffix}` IS NULL".format(
                temp_suffix=temp_suffix, **ref
            )
        ).fetchall()

        assert len(check) == 0, "Some hashes havent been migrated"

        # drop old foreign key, rename, and create new foreign key
        query(
            """
            ALTER TABLE {referencing_table}
            DROP FOREIGN KEY `{constraint_name}`,
            DROP COLUMN `{column_name}`,
            CHANGE COLUMN `{column_name}_{temp_suffix}` `{column_name}`
                {type} DEFAULT NULL
                COMMENT ":blob@{store}:{comment}",
            ADD FOREIGN KEY (`{column_name}`) REFERENCES {ext_table_name}
                (`hash`)
            """.format(
                temp_suffix=temp_suffix,
                ext_table_name=ext.full_table_name,
                type=dj.declare.UUID_DATA_TYPE,
                store=(store or default_store),
                comment=comment,
                **ref
            )
        )

    # Drop the old external table but make sure it's no longer referenced
    # get referencing tables
    refs = [
        {k.lower(): v for k, v in elem.items()}
        for elem in query(
            """
    SELECT concat('`', table_schema, '`.`', table_name, '`') as
        referencing_table, column_name, constraint_name
    FROM information_schema.key_column_usage
    WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
    """.format(
                tab=legacy_external.table_name, db=legacy_external.database
            ),
            as_dict=True,
        ).fetchall()
    ]

    assert not refs, "Some references still exist"

    # drop old external table
    legacy_external.drop_quick()

1	import datajoint as dj	1✔
2	from pathlib import Path	1✔
3	import re	1✔
4	from .utils import user_choice	1✔
5	import logging	1✔
6
7	logger = logging.getLogger(__name__.split(".")[0])	1✔
8
9
10	def migrate_dj011_external_blob_storage_to_dj012(migration_schema, store):	1✔
11	"""
12	Utility function to migrate external blob data from 0.11 to 0.12.
13
14	:param migration_schema: string of target schema to be migrated
15	:param store: string of target dj.config['store'] to be migrated
16	"""
17	if not isinstance(migration_schema, str):	1✔
18	raise ValueError(	1✔
19	"Expected type {} for migration_schema, not {}.".format(
20	str, type(migration_schema)
21	)
22	)
23
24	do_migration = False	×
25	do_migration = (	×
26	user_choice(
27	"""
28	Warning: Ensure the following are completed before proceeding.
29	- Appropriate backups have been taken,
30	- Any existing DJ 0.11.X connections are suspended, and
31	- External config has been updated to new dj.config['stores'] structure.
32	Proceed?
33	""",
34	default="no",
35	)
36	== "yes"
37	)
38	if do_migration:	×
39	_migrate_dj011_blob(dj.Schema(migration_schema), store)	×
40	logger.info(	×
41	"Migration completed for schema: {}, store: {}.".format(
42	migration_schema, store
43	)
44	)
45	return	×
46	logger.warn("No migration performed.")	×
47
48
49	def _migrate_dj011_blob(schema, default_store):	1✔
50	query = schema.connection.query	1✔
51
52	LEGACY_HASH_SIZE = 43	1✔
53
54	legacy_external = dj.FreeTable(	1✔
55	schema.connection, "`{db}`.`~external`".format(db=schema.database)
56	)
57
58	# get referencing tables
59	refs = [	1✔
60	{k.lower(): v for k, v in elem.items()}
61	for elem in query(
62	"""
63	SELECT concat('`', table_schema, '`.`', table_name, '`')
64	as referencing_table, column_name, constraint_name
65	FROM information_schema.key_column_usage
66	WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
67	""".format(
68	tab=legacy_external.table_name, db=legacy_external.database
69	),
70	as_dict=True,
71	).fetchall()
72	]
73
74	for ref in refs:	1✔
75	# get comment
76	column = query(	1✔
77	"SHOW FULL COLUMNS FROM {referencing_table}"
78	'WHERE Field="{column_name}"'.format(**ref),
79	as_dict=True,
80	).fetchone()
81
82	store, comment = re.match(	1✔
83	r":external(-(?P<store>.+))?:(?P<comment>.*)", column["Comment"]
84	).group("store", "comment")
85
86	# get all the hashes from the reference
87	hashes = {	1✔
88	x[0]
89	for x in query(
90	"SELECT `{column_name}` FROM {referencing_table}".format(**ref)
91	)
92	}
93
94	# sanity check make sure that store suffixes match
95	if store is None:	1✔
96	assert all(len(_) == LEGACY_HASH_SIZE for _ in hashes)	1✔
97	else:
98	assert all(_[LEGACY_HASH_SIZE:] == store for _ in hashes)	1✔
99
100	# create new-style external table
101	ext = schema.external[store or default_store]	1✔
102
103	# add the new-style reference field
104	temp_suffix = "tempsub"	1✔
105
106	try:	1✔
107	query(	1✔
108	"""ALTER TABLE {referencing_table}
109	ADD COLUMN `{column_name}_{temp_suffix}` {type} DEFAULT NULL
110	COMMENT ":blob@{store}:{comment}"
111	""".format(
112	type=dj.declare.UUID_DATA_TYPE,
113	temp_suffix=temp_suffix,
114	store=(store or default_store),
115	comment=comment,
116	**ref
117	)
118	)
119	except:	×
120	logger.warn("Column already added")	×
121
122	for _hash, size in zip(*legacy_external.fetch("hash", "size")):	1✔
123	if _hash in hashes:	1✔
124	relative_path = str(Path(schema.database, _hash).as_posix())	1✔
125	uuid = dj.hash.uuid_from_buffer(init_string=relative_path)	1✔
126	external_path = ext._make_external_filepath(relative_path)	1✔
127	if ext.spec["protocol"] == "s3":	1✔
128	contents_hash = dj.hash.uuid_from_buffer(	1✔
129	ext._download_buffer(external_path)
130	)
131	else:
132	contents_hash = dj.hash.uuid_from_file(external_path)	1✔
133	ext.insert1(	1✔
134	dict(
135	filepath=relative_path,
136	size=size,
137	contents_hash=contents_hash,
138	hash=uuid,
139	),
140	skip_duplicates=True,
141	)
142
143	query(	1✔
144	"UPDATE {referencing_table} "
145	"SET `{column_name}_{temp_suffix}`=%s "
146	'WHERE `{column_name}` = "{_hash}"'.format(
147	_hash=_hash, temp_suffix=temp_suffix, **ref
148	),
149	uuid.bytes,
150	)
151
152	# check that all have been copied
153	check = query(	1✔
154	"SELECT * FROM {referencing_table} "
155	"WHERE `{column_name}` IS NOT NULL"
156	" AND `{column_name}_{temp_suffix}` IS NULL".format(
157	temp_suffix=temp_suffix, **ref
158	)
159	).fetchall()
160
161	assert len(check) == 0, "Some hashes havent been migrated"	1✔
162
163	# drop old foreign key, rename, and create new foreign key
164	query(	1✔
165	"""
166	ALTER TABLE {referencing_table}
167	DROP FOREIGN KEY `{constraint_name}`,
168	DROP COLUMN `{column_name}`,
169	CHANGE COLUMN `{column_name}_{temp_suffix}` `{column_name}`
170	{type} DEFAULT NULL
171	COMMENT ":blob@{store}:{comment}",
172	ADD FOREIGN KEY (`{column_name}`) REFERENCES {ext_table_name}
173	(`hash`)
174	""".format(
175	temp_suffix=temp_suffix,
176	ext_table_name=ext.full_table_name,
177	type=dj.declare.UUID_DATA_TYPE,
178	store=(store or default_store),
179	comment=comment,
180	**ref
181	)
182	)
183
184	# Drop the old external table but make sure it's no longer referenced
185	# get referencing tables
186	refs = [	1✔
187	{k.lower(): v for k, v in elem.items()}
188	for elem in query(
189	"""
190	SELECT concat('`', table_schema, '`.`', table_name, '`') as
191	referencing_table, column_name, constraint_name
192	FROM information_schema.key_column_usage
193	WHERE referenced_table_name="{tab}" and referenced_table_schema="{db}"
194	""".format(
195	tab=legacy_external.table_name, db=legacy_external.database
196	),
197	as_dict=True,
198	).fetchall()
199	]
200
201	assert not refs, "Some references still exist"	1✔
202
203	# drop old external table
204	legacy_external.drop_quick()	1✔

datajoint / datajoint-python / #12897

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