tool.objects.dupcheck() - Code Metrics - isomeric/isomer - Measure and Improve Code Quality continuously with Scrutinizer

tool.objects.dupcheck() F
last analyzed 2021-02-24 12:38 UTC

↳ Parent: tool.objects

Complexity

Conditions

Size

Total Lines	183
Code Lines	140

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	39
eloc	140
nop	4
dl	0
loc	183
rs	0
c	0
b	0
f	0

How to fix Long Method Complexity

#!/usr/bin/env python
# -*- coding: UTF-8 -*-

# Isomer - The distributed application framework
# ==============================================
# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.


"""

Module: Objects
===============

Object management functionality and utilities.

"""

from ast import literal_eval
from pprint import pprint

import bson
import deepdiff
import click
from click_didyoumean import DYMGroup

from isomer.logger import error, debug, warn, verbose
from isomer.tool import log, ask, finish
from isomer.tool.database import db


@db.group(
    cls=DYMGroup,
    short_help="Object operations"
)
@click.pass_context
def objects(ctx):
    """[GROUP] Object operations"""
    pass


@objects.command(short_help="modify field values of objects")
@click.option("--schema")
@click.option("--uuid")
@click.option("--filter", "--object-filter")
@click.argument("field")
@click.argument("value")
@click.pass_context
def modify(ctx, schema, uuid, object_filter, field, value):
    """Modify field values of objects"""
    database = ctx.obj["db"]

    model = database.objectmodels[schema]
    obj = None

    if uuid:
        obj = model.find_one({"uuid": uuid})
    elif object_filter:
        obj = model.find_one(literal_eval(object_filter))
    else:
        log("No object uuid or filter specified.", lvl=error)

    if obj is None:
        log("No object found", lvl=error)
        return

    log("Object found, modifying", lvl=debug)
    try:
        new_value = literal_eval(value)
    except ValueError:
        log("Interpreting value as string")
        new_value = str(value)

    obj._fields[field] = new_value
    obj.validate()
    log("Changed object validated", lvl=debug)
    obj.save()
    finish(ctx)


@objects.command(short_help="view objects")
@click.option("--schema", default=None)
@click.option("--uuid", default=None)
@click.option("--object-filter", "--filter", default=None)
@click.pass_context
def view(ctx, schema, uuid, object_filter):
    """Show stored objects"""

    database = ctx.obj["db"]

    if schema is None:
        log("No schema given. Read the help", lvl=warn)
        return

    model = database.objectmodels[schema]

    if uuid:
        obj = model.find({"uuid": uuid})
    elif object_filter:
        obj = model.find(literal_eval(object_filter))
    else:
        obj = model.find()

    if obj is None or model.count() == 0:
        log("No objects found.", lvl=warn)

    for item in obj:
        pprint(item._fields)

    finish(ctx)


@objects.command(short_help="view objects")
@click.option("--schema", default=None)
@click.option("--uuid", default=None)
@click.option("--object-filter", "--filter", default=None)
@click.option(
    "--yes", "-y", help="Assume yes to a safety question", default=False, is_flag=True
)
@click.pass_context
def delete(ctx, schema, uuid, object_filter, yes):
    """Delete stored objects (CAUTION!)"""

    database = ctx.obj["db"]

    if schema is None:
        log("No schema given. Read the help", lvl=warn)
        return

    model = database.objectmodels[schema]

    if uuid:
        count = model.count({"uuid": uuid})
        obj = model.find({"uuid": uuid}, validation=False)
    elif object_filter:
        count = model.count(literal_eval(object_filter))
        obj = model.find(literal_eval(object_filter), validation=False)
    else:
        count = model.count()
        obj = model.find(validation=False)

    if count == 0:
        log("No objects to delete found")
        return

    if not yes and not ask(
        "Are you sure you want to delete %i objects" % count,
        default=False,
        data_type="bool",
        show_hint=True,
    ):
        return

    for item in obj:
        item.delete()

    finish(ctx)


@objects.command(short_help="drop a whole collection of objects")
@click.option("--schema", default=None)
@click.option(
    "--yes", "-y", help="Assume yes to a safety question", default=False, is_flag=True
)
@click.pass_context
def drop(ctx, schema, yes):
    """Delete a whole collection of stored objects (CAUTION!)"""

    database = ctx.obj["db"]

    if schema is None:
        log("No schema given. Read the help", lvl=warn)
        return

    if not yes and not ask(
        "Are you sure you want to drop the whole collection",
        default=False,
        data_type="bool",
        show_hint=True,
    ):
        return

    model = database.objectmodels[schema]
    collection = model.collection()
    collection.drop()

    finish(ctx)


@objects.command(short_help="Validates stored objects")
@click.option("--schema", "-s", default=None, help="Specify object schema to validate")
@click.option(
    "--all-schemata",
    "--all",
    help="Agree to validate all objects, if no schema given",
    is_flag=True,
)
@click.pass_context
def validate(ctx, schema, all_schemata):
    """Validates all objects or all objects of a given schema."""

    database = ctx.obj["db"]

    if schema is None:
        if all_schemata is False:
            log("No schema given. Read the help", lvl=warn)
            return
        else:
            schemata = database.objectmodels.keys()
    else:
        schemata = [schema]

    for schema in schemata:
        try:
            things = database.objectmodels[schema]
            with click.progressbar(
                things.find(), length=things.count(), label="Validating %15s" % schema
            ) as object_bar:
                for obj in object_bar:
                    obj.validate()
        except Exception as e:

            log(
                "Exception while validating:",
                schema,
                e,
                type(e),
                "\n\nFix this object and rerun validation!",
                emitter="MANAGE",
                lvl=error,
            )

    finish(ctx)


@objects.command(short_help="find in object model fields")
@click.option(
    "--search",
    help="Argument to search for in object model fields",
    default=None,
    metavar="<text>",
)
@click.option("--by-type", help="Find all fields by type", default=False, is_flag=True)
@click.option(
    "--obj", default=None, help="Search in specified object model", metavar="<name>"
)
@click.pass_context
def find_field(ctx, search, by_type, obj):
    """Find fields in registered data models."""

    # TODO: Fix this to work recursively on all possible subschemes
    if search is None:
        search = ask("Enter search term")

    database = ctx.obj["db"]

    def find(search_schema, search_field, find_result=None, key=""):
        """Examine a schema to find fields by type or name"""

        if find_result is None:
            find_result = []
        fields = search_schema["properties"]
        if not by_type:
            if search_field in fields:
                find_result.append(key)
                # log("Found queried fieldname in ", model)
        else:
            for field in fields:
                try:
                    if "type" in fields[field]:
                        # log(fields[field], field)
                        if fields[field]["type"] == search_field:
                            find_result.append((key, field))
                            # log("Found field", field, "in", model)
                except KeyError as e:
                    log("Field access error:", e, type(e), exc=True, lvl=debug)

        if "properties" in fields:
            # log('Sub properties checking:', fields['properties'])
            find_result.append(
                find(
                    fields["properties"], search_field, find_result, key=fields["name"]
                )
            )

        for field in fields:
            if "items" in fields[field]:
                if "properties" in fields[field]["items"]:
                    # log('Sub items checking:', fields[field])
                    find_result.append(
                        find(
                            fields[field]["items"], search_field, find_result, key=field
                        )
                    )
                else:
                    pass
                    # log('Items without proper definition!')

        return find_result

    if obj is not None:
        schema = database.objectmodels[obj]._schema
        result = find(schema, search, [], key="top")
        if result:
            # log(args.object, result)
            print(obj)
            pprint(result)
    else:
        for model, thing in database.objectmodels.items():
            schema = thing._schema

            result = find(schema, search, [], key="top")
            if result:
                print(model)
                # log(model, result)
                print(result)

    finish(ctx)


@objects.command(short_help="Find illegal _id fields")
@click.option(
    "--delete-duplicates",
    "--delete",
    default=False,
    is_flag=True,
    help="Delete found duplicates",
)
@click.option(
    "--fix", default=False, is_flag=True, help="Tries to fix faulty object ids"
)
@click.option(
    "--test",
    default=False,
    is_flag=True,
    help="Test if faulty objects have clones with correct ids",
)
@click.option("--schema", default=None, help="Work on specified schema only")
@click.pass_context
def illegalcheck(ctx, schema, delete_duplicates, fix, test):
    """Tool to find erroneous objects created with old legacy bugs. Should be
    obsolete!"""

    database = ctx.obj["db"]

    if delete_duplicates and fix:
        log("Delete and fix operations are exclusive.")
        return

    if schema is None:
        schemata = database.objectmodels.keys()
    else:
        schemata = [schema]

    for thing in schemata:
        log("Schema:", thing)
        for item in database.objectmodels[thing].find():
            if not isinstance(item._fields["_id"], bson.objectid.ObjectId):
                if not delete_duplicates:
                    log(item.uuid)
                    log(item._fields, pretty=True, lvl=verbose)
                if test:
                    if database.objectmodels[thing].count({"uuid": item.uuid}) == 1:
                        log("Only a faulty object exists.")
                if delete_duplicates:
                    item.delete()
                if fix:
                    _id = item._fields["_id"]
                    item._fields["_id"] = bson.objectid.ObjectId(_id)
                    if not isinstance(item._fields["_id"], bson.objectid.ObjectId):
                        log("Object mongo ID field not valid!", lvl=warn)
                    item.save()
                    database.objectmodels[thing].find_one({"_id": _id}).delete()
    finish(ctx)


@objects.command(short_help="Find duplicates by UUID")
@click.option(
    "--delete-duplicates",
    "--delete",
    default=False,
    is_flag=True,
    help="Delete found duplicates",
)
@click.option(
    "--do-merge", "--merge", default=False, is_flag=True, help="Merge found duplicates"
)
@click.option("--schema", default=None, help="Work on specified schema only")
@click.pass_context
def dupcheck(ctx, delete_duplicates, do_merge, schema):
    """Tool to check for duplicate objects. Which should never happen."""

    def handle_schema(check_schema):
        dupes = {}
        dupe_count = 0
        count = 0

        for item in database.objectmodels[check_schema].find():
            if item.uuid in dupes:
                dupes[item.uuid].append(item)
                dupe_count += 1
            else:
                dupes[item.uuid] = [item]
            count += 1

        if len(dupes) > 0:
            log(
                dupe_count,
                "duplicates of",
                count,
                "items total of type",
                check_schema,
                "found:",
            )
            log(dupes.keys(), pretty=True, lvl=verbose)

        if delete_duplicates:
            log("Deleting duplicates")
            for item in dupes:
                database.objectmodels[check_schema].find_one({"uuid": item}).delete()

            log("Done for schema", check_schema)
        elif do_merge:

            def merge(a, b, path=None):
                """merges b into a"""

                if path is None:
                    path = []
                for key in b:
                    if key in a:
                        if isinstance(a[key], dict) and isinstance(b[key], dict):
                            merge(a[key], b[key], path + [str(key)])
                        elif a[key] == b[key]:
                            pass  # same leaf value
                        else:
                            log("Conflict at", path, key, ":", a[key], "<->", b[key])
                            resolve = ""
                            while resolve not in ("a", "b"):
                                resolve = ask("Choose? (a or b)")
                            if resolve == "a":
                                b[key] = a[key]
                            else:
                                a[key] = b[key]
                    else:
                        a[key] = b[key]
                return a

            log(dupes, pretty=True, lvl=verbose)

            for item in dupes:
                if len(dupes[item]) == 1:
                    continue
                ignore = False
                while len(dupes[item]) > 1 or ignore is False:
                    log(len(dupes[item]), "duplicates found:")
                    for index, dupe in enumerate(dupes[item]):
                        log("Candidate #", index, ":")
                        log(dupe._fields, pretty=True)
                    request = ask("(d)iff, (m)erge, (r)emove, (i)gnore, (q)uit?")
                    if request == "q":
                        log("Done")
                        return
                    elif request == "i":
                        ignore = True
                        break
                    elif request == "r":
                        delete_request = -2
                        while delete_request == -2 or -1 > delete_request > len(
                            dupes[item]
                        ):
                            delete_request = ask(
                                "Which one? (0-%i or -1 to cancel)"
                                % (len(dupes[item]) - 1),
                                data_type="int",
                            )
                        if delete_request == -1:
                            continue
                        else:
                            log("Deleting candidate #", delete_request)
                            dupes[item][delete_request].delete()
                            break
                    elif request in ("d", "m"):
                        merge_request_a = -2
                        merge_request_b = -2

                        while merge_request_a == -2 or -1 > merge_request_a > len(
                            dupes[item]
                        ):
                            merge_request_a = ask(
                                "Merge from? (0-%i or -1 to cancel)"
                                % (len(dupes[item]) - 1),
                                data_type="int",
                            )
                        if merge_request_a == -1:
                            continue

                        while merge_request_b == -2 or -1 > merge_request_b > len(
                            dupes[item]
                        ):
                            merge_request_b = ask(
                                "Merge into? (0-%i or -1 to cancel)"
                                % (len(dupes[item]) - 1),
                                data_type="int",
                            )
                        if merge_request_b == -1:
                            continue

                        log(
                            deepdiff.DeepDiff(
                                dupes[item][merge_request_a]._fields,
                                dupes[item][merge_request_b]._fields,
                            ),
                            pretty=True,
                        )

                        if request == "m":
                            log(
                                "Merging candidates",
                                merge_request_a,
                                "and",
                                merge_request_b,
                            )

                            _id = dupes[item][merge_request_b]._fields["_id"]
                            if not isinstance(_id, bson.objectid.ObjectId):
                                _id = bson.objectid.ObjectId(_id)

                            dupes[item][merge_request_a]._fields["_id"] = _id
                            merge(
                                dupes[item][merge_request_b]._fields,
                                dupes[item][merge_request_a]._fields,
                            )

                            log(
                                "Candidate after merge:",
                                dupes[item][merge_request_b]._fields,
                                pretty=True,
                            )

                            store = ""
                            while store not in ("n", "y"):
                                store = ask("Store?")
                            if store == "y":
                                dupes[item][merge_request_b].save()
                                dupes[item][merge_request_a].delete()
                                break

    database = ctx.obj["db"]

    if schema is None:
        schemata = database.objectmodels.keys()
    else:
        schemata = [schema]

    for thing in schemata:
        handle_schema(thing)

    finish(ctx)


1			#!/usr/bin/env python
2			# -- coding: UTF-8 --
3
4			# Isomer - The distributed application framework
5			# ==============================================
6			# Copyright (C) 2011-2020 Heiko 'riot' Weinen <[email protected]> and others.
7			#
8			# This program is free software: you can redistribute it and/or modify
9			# it under the terms of the GNU Affero General Public License as published by
10			# the Free Software Foundation, either version 3 of the License, or
11			# (at your option) any later version.
12			#
13			# This program is distributed in the hope that it will be useful,
14			# but WITHOUT ANY WARRANTY; without even the implied warranty of
15			# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16			# GNU Affero General Public License for more details.
17			#
18			# You should have received a copy of the GNU Affero General Public License
19			# along with this program. If not, see <http://www.gnu.org/licenses/>.
20
21
22			"""
23
24			Module: Objects
25			===============
26
27			Object management functionality and utilities.
28
29			"""
30
31			from ast import literal_eval
32			from pprint import pprint
33
34			import bson
35			import deepdiff
36			import click
37			from click_didyoumean import DYMGroup
38
39			from isomer.logger import error, debug, warn, verbose
40			from isomer.tool import log, ask, finish
41			from isomer.tool.database import db
42
43
44			@db.group(
45			cls=DYMGroup,
46			short_help="Object operations"
47			)
48			@click.pass_context
49			def objects(ctx):
50			"""[GROUP] Object operations"""
51			pass
52
53
54			@objects.command(short_help="modify field values of objects")
55			@click.option("--schema")
56			@click.option("--uuid")
57			@click.option("--filter", "--object-filter")
58			@click.argument("field")
59			@click.argument("value")
60			@click.pass_context
61			def modify(ctx, schema, uuid, object_filter, field, value):
62			"""Modify field values of objects"""
63			database = ctx.obj["db"]
64
65			model = database.objectmodels[schema]
66			obj = None
67
68			if uuid:
69			obj = model.find_one({"uuid": uuid})
70			elif object_filter:
71			obj = model.find_one(literal_eval(object_filter))
72			else:
73			log("No object uuid or filter specified.", lvl=error)
74
75			if obj is None:
76			log("No object found", lvl=error)
77			return
78
79			log("Object found, modifying", lvl=debug)
80			try:
81			new_value = literal_eval(value)
82			except ValueError:
83			log("Interpreting value as string")
84			new_value = str(value)
85
86			obj._fields[field] = new_value
87			obj.validate()
88			log("Changed object validated", lvl=debug)
89			obj.save()
90			finish(ctx)
91
92
93			@objects.command(short_help="view objects")
94			@click.option("--schema", default=None)
95			@click.option("--uuid", default=None)
96			@click.option("--object-filter", "--filter", default=None)
97			@click.pass_context
98			def view(ctx, schema, uuid, object_filter):
99			"""Show stored objects"""
100
101			database = ctx.obj["db"]
102
103			if schema is None:
104			log("No schema given. Read the help", lvl=warn)
105			return
106
107			model = database.objectmodels[schema]
108
109			if uuid:
110			obj = model.find({"uuid": uuid})
111			elif object_filter:
112			obj = model.find(literal_eval(object_filter))
113			else:
114			obj = model.find()
115
116			if obj is None or model.count() == 0:
117			log("No objects found.", lvl=warn)
118
119			for item in obj:
120			pprint(item._fields)
121
122			finish(ctx)
123
124
125			@objects.command(short_help="view objects")
126			@click.option("--schema", default=None)
127			@click.option("--uuid", default=None)
128			@click.option("--object-filter", "--filter", default=None)
129			@click.option(
130			"--yes", "-y", help="Assume yes to a safety question", default=False, is_flag=True
131			)
132			@click.pass_context
133			def delete(ctx, schema, uuid, object_filter, yes):
134			"""Delete stored objects (CAUTION!)"""
135
136			database = ctx.obj["db"]
137
138			if schema is None:
139			log("No schema given. Read the help", lvl=warn)
140			return
141
142			model = database.objectmodels[schema]
143
144			if uuid:
145			count = model.count({"uuid": uuid})
146			obj = model.find({"uuid": uuid}, validation=False)
147			elif object_filter:
148			count = model.count(literal_eval(object_filter))
149			obj = model.find(literal_eval(object_filter), validation=False)
150			else:
151			count = model.count()
152			obj = model.find(validation=False)
153
154			if count == 0:
155			log("No objects to delete found")
156			return
157
158			if not yes and not ask(
159			"Are you sure you want to delete %i objects" % count,
160			default=False,
161			data_type="bool",
162			show_hint=True,
163			):
164			return
165
166			for item in obj:
167			item.delete()
168
169			finish(ctx)
170
171
172			@objects.command(short_help="drop a whole collection of objects")
173			@click.option("--schema", default=None)
174			@click.option(
175			"--yes", "-y", help="Assume yes to a safety question", default=False, is_flag=True
176			)
177			@click.pass_context
178			def drop(ctx, schema, yes):
179			"""Delete a whole collection of stored objects (CAUTION!)"""
180
181			database = ctx.obj["db"]
182
183			if schema is None:
184			log("No schema given. Read the help", lvl=warn)
185			return
186
187			if not yes and not ask(
188			"Are you sure you want to drop the whole collection",
189			default=False,
190			data_type="bool",
191			show_hint=True,
192			):
193			return
194
195			model = database.objectmodels[schema]
196			collection = model.collection()
197			collection.drop()
198
199			finish(ctx)
200
201
202			@objects.command(short_help="Validates stored objects")
203			@click.option("--schema", "-s", default=None, help="Specify object schema to validate")
204			@click.option(
205			"--all-schemata",
206			"--all",
207			help="Agree to validate all objects, if no schema given",
208			is_flag=True,
209			)
210			@click.pass_context
211			def validate(ctx, schema, all_schemata):
212			"""Validates all objects or all objects of a given schema."""
213
214			database = ctx.obj["db"]
215
216			if schema is None:
217			if all_schemata is False:
218			log("No schema given. Read the help", lvl=warn)
219			return
220			else:
221			schemata = database.objectmodels.keys()
222			else:
223			schemata = [schema]
224
225			for schema in schemata:
226			try:
227			things = database.objectmodels[schema]
228			with click.progressbar(
229			things.find(), length=things.count(), label="Validating %15s" % schema
230			) as object_bar:
231			for obj in object_bar:
232			obj.validate()
233			except Exception as e:
234
235			log(
236			"Exception while validating:",
237			schema,
238			e,
239			type(e),
240			"\n\nFix this object and rerun validation!",
241			emitter="MANAGE",
242			lvl=error,
243			)
244
245			finish(ctx)
246
247
248			@objects.command(short_help="find in object model fields")
249			@click.option(
250			"--search",
251			help="Argument to search for in object model fields",
252			default=None,
253			metavar="<text>",
254			)
255			@click.option("--by-type", help="Find all fields by type", default=False, is_flag=True)
256			@click.option(
257			"--obj", default=None, help="Search in specified object model", metavar="<name>"
258			)
259			@click.pass_context
260			def find_field(ctx, search, by_type, obj):
261			"""Find fields in registered data models."""
262
263			# TODO: Fix this to work recursively on all possible subschemes
264			if search is None:
265			search = ask("Enter search term")
266
267			database = ctx.obj["db"]
268
269			def find(search_schema, search_field, find_result=None, key=""):
270			"""Examine a schema to find fields by type or name"""
271
272			if find_result is None:
273			find_result = []
274			fields = search_schema["properties"]
275			if not by_type:
276			if search_field in fields:
277			find_result.append(key)
278			# log("Found queried fieldname in ", model)
279			else:
280			for field in fields:
281			try:
282			if "type" in fields[field]:
283			# log(fields[field], field)
284			if fields[field]["type"] == search_field:
285			find_result.append((key, field))
286			# log("Found field", field, "in", model)
287			except KeyError as e:
288			log("Field access error:", e, type(e), exc=True, lvl=debug)
289
290			if "properties" in fields:
291			# log('Sub properties checking:', fields['properties'])
292			find_result.append(
293			find(
294			fields["properties"], search_field, find_result, key=fields["name"]
295			)
296			)
297
298			for field in fields:
299			if "items" in fields[field]:
300			if "properties" in fields[field]["items"]:
301			# log('Sub items checking:', fields[field])
302			find_result.append(
303			find(
304			fields[field]["items"], search_field, find_result, key=field
305			)
306			)
307			else:
308			pass
309			# log('Items without proper definition!')
310
311			return find_result
312
313			if obj is not None:
314			schema = database.objectmodels[obj]._schema
315			result = find(schema, search, [], key="top")
316			if result:
317			# log(args.object, result)
318			print(obj)
319			pprint(result)
320			else:
321			for model, thing in database.objectmodels.items():
322			schema = thing._schema
323
324			result = find(schema, search, [], key="top")
325			if result:
326			print(model)
327			# log(model, result)
328			print(result)
329
330			finish(ctx)
331
332
333			@objects.command(short_help="Find illegal _id fields")
334			@click.option(
335			"--delete-duplicates",
336			"--delete",
337			default=False,
338			is_flag=True,
339			help="Delete found duplicates",
340			)
341			@click.option(
342			"--fix", default=False, is_flag=True, help="Tries to fix faulty object ids"
343			)
344			@click.option(
345			"--test",
346			default=False,
347			is_flag=True,
348			help="Test if faulty objects have clones with correct ids",
349			)
350			@click.option("--schema", default=None, help="Work on specified schema only")
351			@click.pass_context
352			def illegalcheck(ctx, schema, delete_duplicates, fix, test):
353			"""Tool to find erroneous objects created with old legacy bugs. Should be
354			obsolete!"""
355
356			database = ctx.obj["db"]
357
358			if delete_duplicates and fix:
359			log("Delete and fix operations are exclusive.")
360			return
361
362			if schema is None:
363			schemata = database.objectmodels.keys()
364			else:
365			schemata = [schema]
366
367			for thing in schemata:
368			log("Schema:", thing)
369			for item in database.objectmodels[thing].find():
370			if not isinstance(item._fields["_id"], bson.objectid.ObjectId):
371			if not delete_duplicates:
372			log(item.uuid)
373			log(item._fields, pretty=True, lvl=verbose)
374			if test:
375			if database.objectmodels[thing].count({"uuid": item.uuid}) == 1:
376			log("Only a faulty object exists.")
377			if delete_duplicates:
378			item.delete()
379			if fix:
380			_id = item._fields["_id"]
381			item._fields["_id"] = bson.objectid.ObjectId(_id)
382			if not isinstance(item._fields["_id"], bson.objectid.ObjectId):
383			log("Object mongo ID field not valid!", lvl=warn)
384			item.save()
385			database.objectmodels[thing].find_one({"_id": _id}).delete()
386			finish(ctx)
387
388
389			@objects.command(short_help="Find duplicates by UUID")
390			@click.option(
391			"--delete-duplicates",
392			"--delete",
393			default=False,
394			is_flag=True,
395			help="Delete found duplicates",
396			)
397			@click.option(
398			"--do-merge", "--merge", default=False, is_flag=True, help="Merge found duplicates"
399			)
400			@click.option("--schema", default=None, help="Work on specified schema only")
401			@click.pass_context
402			def dupcheck(ctx, delete_duplicates, do_merge, schema):
403			"""Tool to check for duplicate objects. Which should never happen."""
404
405			def handle_schema(check_schema):
406			dupes = {}
407			dupe_count = 0
408			count = 0
409
410			for item in database.objectmodels[check_schema].find():
411			if item.uuid in dupes:
412			dupes[item.uuid].append(item)
413			dupe_count += 1
414			else:
415			dupes[item.uuid] = [item]
416			count += 1
417
418			if len(dupes) > 0:
419			log(
420			dupe_count,
421			"duplicates of",
422			count,
423			"items total of type",
424			check_schema,
425			"found:",
426			)
427			log(dupes.keys(), pretty=True, lvl=verbose)
428
429			if delete_duplicates:
430			log("Deleting duplicates")
431			for item in dupes:
432			database.objectmodels[check_schema].find_one({"uuid": item}).delete()
433
434			log("Done for schema", check_schema)
435			elif do_merge:
436
437			def merge(a, b, path=None):
438			"""merges b into a"""
439
440			if path is None:
441			path = []
442			for key in b:
443			if key in a:
444			if isinstance(a[key], dict) and isinstance(b[key], dict):
445			merge(a[key], b[key], path + [str(key)])
446			elif a[key] == b[key]:
447			pass # same leaf value
448			else:
449			log("Conflict at", path, key, ":", a[key], "<->", b[key])
450			resolve = ""
451			while resolve not in ("a", "b"):
452			resolve = ask("Choose? (a or b)")
453			if resolve == "a":
454			b[key] = a[key]
455			else:
456			a[key] = b[key]
457			else:
458			a[key] = b[key]
459			return a
460
461			log(dupes, pretty=True, lvl=verbose)
462
463			for item in dupes:
464			if len(dupes[item]) == 1:
465			continue
466			ignore = False
467			while len(dupes[item]) > 1 or ignore is False:
468			log(len(dupes[item]), "duplicates found:")
469			for index, dupe in enumerate(dupes[item]):
470			log("Candidate #", index, ":")
471			log(dupe._fields, pretty=True)
472			request = ask("(d)iff, (m)erge, (r)emove, (i)gnore, (q)uit?")
473			if request == "q":
474			log("Done")
475			return
476			elif request == "i":
477			ignore = True
478			break
479			elif request == "r":
480			delete_request = -2
481			while delete_request == -2 or -1 > delete_request > len(
482			dupes[item]
483			):
484			delete_request = ask(
485			"Which one? (0-%i or -1 to cancel)"
486			% (len(dupes[item]) - 1),
487			data_type="int",
488			)
489			if delete_request == -1:
490			continue
491			else:
492			log("Deleting candidate #", delete_request)
493			dupes[item][delete_request].delete()
494			break
495			elif request in ("d", "m"):
496			merge_request_a = -2
497			merge_request_b = -2
498
499			while merge_request_a == -2 or -1 > merge_request_a > len(
500			dupes[item]
501			):
502			merge_request_a = ask(
503			"Merge from? (0-%i or -1 to cancel)"
504			% (len(dupes[item]) - 1),
505			data_type="int",
506			)
507			if merge_request_a == -1:
508			continue
509
510			while merge_request_b == -2 or -1 > merge_request_b > len(
511			dupes[item]
512			):
513			merge_request_b = ask(
514			"Merge into? (0-%i or -1 to cancel)"
515			% (len(dupes[item]) - 1),
516			data_type="int",
517			)
518			if merge_request_b == -1:
519			continue
520
521			log(
522			deepdiff.DeepDiff(
523			dupes[item][merge_request_a]._fields,
524			dupes[item][merge_request_b]._fields,
525			),
526			pretty=True,
527			)
528
529			if request == "m":
530			log(
531			"Merging candidates",
532			merge_request_a,
533			"and",
534			merge_request_b,
535			)
536
537			_id = dupes[item][merge_request_b]._fields["_id"]
538			if not isinstance(_id, bson.objectid.ObjectId):
539			_id = bson.objectid.ObjectId(_id)
540
541			dupes[item][merge_request_a]._fields["_id"] = _id
542			merge(
543			dupes[item][merge_request_b]._fields,
544			dupes[item][merge_request_a]._fields,
545			)
546
547			log(
548			"Candidate after merge:",
549			dupes[item][merge_request_b]._fields,
550			pretty=True,
551			)
552
553			store = ""
554			while store not in ("n", "y"):
555			store = ask("Store?")
556			if store == "y":
557			dupes[item][merge_request_b].save()
558			dupes[item][merge_request_a].delete()
559			break
560
561			database = ctx.obj["db"]
562
563			if schema is None:
564			schemata = database.objectmodels.keys()
565			else:
566			schemata = [schema]
567
568			for thing in schemata:
569			handle_schema(thing)
570
571			finish(ctx)
572

isomeric / isomer

tool.objects.dupcheck() F last analyzed 2021-02-24 12:38 UTC

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

tool.objects.dupcheck() F
last analyzed 2021-02-24 12:38 UTC