-
Notifications
You must be signed in to change notification settings - Fork 11
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
13 changed files
with
248 additions
and
48 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,63 @@ | ||
| """cascades | ||
| Revision ID: 3e9675b35dfc | ||
| Revises: cb568ec81000 | ||
| Create Date: 2020-08-12 14:48:35.592316 | ||
| """ | ||
|
|
||
| # revision identifiers, used by Alembic. | ||
| revision = '3e9675b35dfc' | ||
| down_revision = 'cb568ec81000' | ||
|
|
||
| from alembic import op | ||
|
|
||
| is_sqlite = op.get_bind().dialect.name == 'sqlite' | ||
| convention = ({"fk": "FK_%(table_name)s_%(referred_table_name)s"} if is_sqlite else | ||
| {"fk": "%(table_name)s_%(column_0_name)s_fkey"}) | ||
|
|
||
|
|
||
| def upgrade(): | ||
|
|
||
| args = [ | ||
| ('conceptscheme_source', 'conceptscheme', 'conceptscheme_id', 'id'), | ||
| ('conceptscheme_source', 'source', 'source_id', 'id'), | ||
| ('concept_source', 'concept', 'concept_id', 'id'), | ||
| ('concept_source', 'source', 'source_id', 'id'), | ||
| ('conceptscheme_language', 'conceptscheme', 'conceptscheme_id', 'id'), | ||
| ('match', 'concept', 'concept_id', 'id'), | ||
| ('label', 'language', 'language_id', 'id'), | ||
| ('note', 'language', 'language_id', 'id'), | ||
| ('collection_concept', 'concept', 'concept_id', 'id'), | ||
| ('collection_concept', 'concept', 'collection_id', 'id'), | ||
| ('conceptscheme_note', 'conceptscheme', 'conceptscheme_id', 'id'), | ||
| ('conceptscheme_label', 'conceptscheme', 'conceptscheme_id', 'id'), | ||
| ('visitation', 'concept', 'concept_id', 'id'), | ||
| ('visitation', 'conceptscheme', 'conceptscheme_id', 'id'), | ||
| ('concept_note', 'concept', 'concept_id', 'id'), | ||
| ('concept_note', 'note', 'note_id', 'id'), | ||
| ('concept_related_concept', 'concept', 'concept_id_from', 'id'), | ||
| ('concept_related_concept', 'concept', 'concept_id_to', 'id'), | ||
| ('concept_hierarchy_concept', 'concept', 'concept_id_narrower', 'id'), | ||
| ('concept_hierarchy_concept', 'concept', 'concept_id_broader', 'id'), | ||
| ('concept_hierarchy_collection', 'concept', 'concept_id_broader', 'id'), | ||
| ('concept_hierarchy_collection', 'concept', 'collection_id_narrower', 'id'), | ||
| ('concept_label', 'concept', 'concept_id', 'id'), | ||
| ('concept_label', 'label', 'label_id', 'id'), | ||
| ] | ||
|
|
||
| for source_table, referent_table, local_col, remote_col in args: | ||
| with op.batch_alter_table(source_table, naming_convention=convention) as batch_op: | ||
| constraint_name = convention['fk'] % { | ||
| 'table_name': source_table, | ||
| 'referred_table_name': referent_table, | ||
| 'column_0_name': local_col | ||
| } | ||
| batch_op.drop_constraint(constraint_name) | ||
| batch_op.create_foreign_key( | ||
| constraint_name, referent_table, | ||
| [local_col], [remote_col], ondelete='cascade') | ||
|
|
||
|
|
||
| def downgrade(): | ||
| pass |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,95 @@ | ||
| # -*- coding: utf-8 -*- | ||
| import argparse | ||
| import logging | ||
|
|
||
| from pyramid.paster import get_appsettings | ||
| from pyramid.paster import setup_logging | ||
| from pytz import timezone | ||
| from sqlalchemy import engine_from_config | ||
|
|
||
| try: | ||
| from builtins import input | ||
| except ImportError: | ||
| input = raw_input | ||
|
|
||
| timezone_brussels = timezone('Europe/Brussels') | ||
| log = logging.getLogger(__name__) | ||
|
|
||
|
|
||
| def delete_scheme(settings, scheme_id): | ||
| engine = engine_from_config(settings, 'sqlalchemy.') | ||
| with engine.connect() as con: | ||
| concept_ids = con.execute( | ||
| 'select id from concept where conceptscheme_id={}'.format(scheme_id) | ||
| ) | ||
| for row in concept_ids: | ||
| concept_id = row[0] | ||
| delete_concept(concept_id, con) | ||
|
|
||
| con.execute('delete from note where note.id in' | ||
| '(select note_id from conceptscheme_note ' | ||
| 'where conceptscheme_id={})'.format(scheme_id)) | ||
| con.execute('delete from source where source.id in' | ||
| '(select source_id from conceptscheme_source ' | ||
| 'where conceptscheme_id={})'.format(scheme_id)) | ||
| con.execute('delete from label where label.id in' | ||
| '(select label_id from conceptscheme_label ' | ||
| 'where conceptscheme_id={})'.format(scheme_id)) | ||
| con.execute('delete from conceptscheme where id = {}'.format(scheme_id)) | ||
|
|
||
|
|
||
| def delete_concept(concept_id, con): | ||
| con.execute('delete from note where note.id in' | ||
| '(select note_id from concept_note where concept_id={})' | ||
| .format(concept_id)) | ||
| con.execute('delete from source where source.id in' | ||
| '(select source_id from concept_source ' | ||
| 'where concept_id={})'.format(concept_id)) | ||
| con.execute('delete from label where label.id in' | ||
| '(select label_id from concept_label ' | ||
| 'where concept_id={})'.format(concept_id)) | ||
| delete_child_concepts(concept_id, con) | ||
| con.execute('delete from concept where id = {}'.format(concept_id)) | ||
|
|
||
|
|
||
| def delete_child_concepts(concept_id, con): | ||
| select_children = ('select collection_id_narrower as child ' | ||
| 'from concept_hierarchy_collection ' | ||
| 'where concept_id_broader = {0} ' | ||
| 'union ' | ||
| 'select concept_id_narrower as child ' | ||
| 'from concept_hierarchy_concept ' | ||
| 'where concept_id_broader = {0};'.format(concept_id)) | ||
| children = con.execute(select_children) | ||
| for row in children: | ||
| child_id = row[0] | ||
| delete_concept(child_id, con) | ||
|
|
||
|
|
||
| def main(): | ||
| parser = argparse.ArgumentParser( | ||
| description="Delete a conceptscheme. ", | ||
| usage="remove_schema development.ini --id=1") | ||
| parser.add_argument('settings_file', | ||
| help="<The location of the settings file>#<app-name>") | ||
| parser.add_argument("--id", type=int, required=True, | ||
| help="the conceptscheme id") | ||
| parser.add_argument("--no-input", action='store_true', | ||
| help="Don't stop script for user input") | ||
| args, _ = parser.parse_known_args() | ||
|
|
||
| config_uri = args.settings_file | ||
| setup_logging(config_uri) | ||
| settings = get_appsettings(config_uri) | ||
| id = args.id | ||
| print( | ||
| "The conceptscheme with id {} will be deleted".format(id) | ||
| ) | ||
| if not args.no_input: | ||
| input("Press [Enter] to continue.") | ||
|
|
||
| delete_scheme(settings, id) | ||
|
|
||
|
|
||
| if __name__ == '__main__': # pragma: no cover | ||
| main() |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| import os | ||
|
|
||
| from paste.deploy import appconfig | ||
| from skosprovider_sqlalchemy.models import Collection | ||
| from skosprovider_sqlalchemy.models import Concept | ||
| from skosprovider_sqlalchemy.models import ConceptScheme | ||
| from skosprovider_sqlalchemy.models import Label | ||
| from skosprovider_sqlalchemy.models import LabelType | ||
| from skosprovider_sqlalchemy.models import Language | ||
| from skosprovider_sqlalchemy.models import Note | ||
| from skosprovider_sqlalchemy.models import NoteType | ||
| from skosprovider_sqlalchemy.models import Source | ||
| from skosprovider_sqlalchemy.models import Visitation | ||
|
|
||
| from atramhasis.scripts import delete_scheme | ||
| from tests import DbTest | ||
| from tests import db_session | ||
| from tests import fill_db | ||
| from tests import setup_db | ||
|
|
||
| TEST_DIR = os.path.dirname(__file__) | ||
| settings = appconfig('config:' + os.path.join(TEST_DIR, 'conf_test.ini')) | ||
|
|
||
|
|
||
| def setUpModule(): | ||
| setup_db() | ||
| fill_db() | ||
|
|
||
|
|
||
| class DeleteSchemeTest(DbTest): | ||
|
|
||
| def test_delete(self): | ||
| with db_session() as session: | ||
| for id in range(1, 10): | ||
| delete_scheme.delete_scheme(settings, id) | ||
| assert len(session.query(ConceptScheme).all()) == 0 | ||
| assert len(session.query(Concept).all()) == 0 | ||
| assert len(session.query(Collection).all()) == 0 | ||
| assert len(session.query(Note).all()) == 0 | ||
| assert len(session.query(Source).all()) == 0 | ||
| assert len(session.query(Visitation).all()) == 0 | ||
| assert len(session.query(Label).all()) == 0 | ||
| assert len(session.query(LabelType).all()) != 0 | ||
| assert len(session.query(NoteType).all()) != 0 | ||
| assert len(session.query(Language).all()) != 0 | ||
|
|
Oops, something went wrong.