senaite.core.migration.migrator.ATDXContentMigrator.migrate() - Code Metrics - Inspection of "Converted Container to Dexterity Contents" - senaite/senaite.core - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Pull Request — 2.x (#1911)

by Jordi

created 2022-03-30 16:12 UTC

ATDXContentMigrator.migrate() A

↳ Parent: senaite.core.migration.migrator

Complexity

Conditions

Size

Total Lines	39
Code Lines	14

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	14
dl	0
loc	39
rs	9.7
c	0
b	0
f	0
cc	3
nop	3

# -*- coding: utf-8 -*-

import json

import six

from Acquisition import aq_parent
from bika.lims import api
from bika.lims import logger
from bika.lims.interfaces import IAuditable
from plone.dexterity.interfaces import IDexterityContent
from Products.Archetypes.interfaces import IBaseObject
from Products.Archetypes.interfaces import IField
from senaite.core.interfaces import IContentMigrator
from senaite.core.interfaces import IFieldMigrator
from senaite.core.migration.utils import copyPermMap
from z3c.form.interfaces import IDataManager
from zope.component import adapts
from zope.component import getMultiAdapter
from zope.interface import alsoProvides
from zope.interface import directlyProvidedBy
from zope.interface import implementer

SKIP_FIELDS = [
    "id",
    "allowDiscussion",
    "subject",
    "location",
    "contributors",
    "creators",
    "effectiveDate",
    "expirationDate",
    "language",
    "rights",
    "creation_date",
]


@implementer(IContentMigrator)
class ContentMigrator(object):
    """SENAITE content migrator
    """
    def __init__(self, src, target):
        self.src = src
        self.target = target

    def migrate(self):
        """Run the migration
        """
        raise NotImplementedError("Must be implemented by subclass")

    def uncatalog_object(self, obj):
        """Uncatalog the object for all catalogs
        """
        # uncatalog from registered catalogs
        obj.unindexObject()
        # explicitly uncatalog from uid_catalog
        uid_catalog = api.get_tool("uid_catalog")
        url = "/".join(obj.getPhysicalPath()[2:])
        uid_catalog.uncatalog_object(url)

    def catalog_object(self, obj):
        """Catalog the object
        """
        obj.reindexObject()

    def copy_uid(self, obj, uid):
        """Set uid on object
        """
        if api.is_dexterity_content(obj):
            setattr(obj, "_plone.uuid", uid)
        elif api.is_at_content(obj):
            setattr(obj, "_at_uid", uid)
        else:
            raise TypeError("Cannot set UID on that object")

    def copy_dates(self, src, target):
        """copy modification/creation date
        """
        created = api.get_creation_date(src)
        modified = api.get_modification_date(src)
        target.creation_date = created
        target.setModificationDate(modified)

    def copy_creators(self, src, target):
        """Copy creators
        """
        target.setCreators(src.listCreators())

    def copy_workflow_history(self, src, target):
        """Copy workflow history
        """
        wfh = getattr(src, "workflow_history", None)
        if wfh:
            wfh = copyPermMap(wfh)
            target.workflow_history = wfh

    def copy_marker_interfaces(self, src, target):
        """Copy marker interfaces
        """
        alsoProvides(target, directlyProvidedBy(src))

    def copy_snapshots(self, src, target):
        """copy over snapshots from source -> target
        """
        snapshots = api.snapshot.get_snapshots(src)
        storage = api.snapshot.get_storage(target)
        storage[:] = map(json.dumps, snapshots)[:]
        alsoProvides(target, IAuditable)

    def delete_object(self, obj):
        """delete the object w/o firing events
        """
        self.uncatalog_object(obj)
        parent = aq_parent(obj)
        parent._delObject(obj.getId(), suppress_events=True)

    def copy_fields(self, src, target, mapping):
        """Copy fields
        """
        src_fields = api.get_fields(src)
        for fname, field in src_fields.items():
            if fname in SKIP_FIELDS:
                continue
            field_migrator = getMultiAdapter(
                (field, src, target), interface=IFieldMigrator)
            # migrate the field
            field_migrator.migrate(mapping)


class ATDXContentMigrator(ContentMigrator):
    """Migrate from AT to DX contents
    """
    adapts(IBaseObject, IDexterityContent)

    def migrate(self, mapping=None, delete_src=True):
        """Migrate AT content to DX

        :param mapping: a mapping from source schema field name to a tuple of
                        (accessor name, target field name, default value)
        """
        if mapping is None:
            mapping = {}

        # copy_fields
        self.copy_fields(self.src, self.target, mapping)

        # copy the UID
        self.copy_uid(self.src, self.target)

        # copy auditlog
        self.copy_snapshots(self.src, self.target)

        # copy creators
        self.copy_creators(self.src, self.target)

        # copy workflow history
        self.copy_workflow_history(self.src, self.target)

        # copy marker interfaces
        self.copy_marker_interfaces(self.src, self.target)

        # copy dates
        self.copy_dates(self.src, self.target)

        # uncatalog the source object
        self.uncatalog_object(self.src)

        # reindex the new object
        self.catalog_object(self.target)

        # delete source object if requested
        if delete_src:
            self.delete_object(self.src)


@implementer(IFieldMigrator)
class FieldMigrator(object):
    """SENAITE field migrator
    """
    def __init__(self, field, src, target):
        self.field = field
        self.src = src
        self.target = target

    def migrate(self):
        raise NotImplementedError("Must be implemented by subclass")


class ATDXFieldMigrator(FieldMigrator):
    """SENAITE AT to DX field migrator
    """
    adapts(IField, IBaseObject, IDexterityContent)

    def migrate(self, mapping):
        # get all fields on the target
        target_fields = api.get_fields(self.target)

        fieldname = self.field.getName()

        # check if we have a mapping for this field
        accessor, target_fieldname, default = (None, None, None, )
        if fieldname in mapping:
            accessor, target_fieldname, default = mapping[fieldname]

        target_field = None
        if target_fieldname:
            # get the target field with the mapped name
            target_field = target_fields.get(target_fieldname)
        else:
            # check for a field with the same name on the target
            target_field = target_fields.get(fieldname)

        # no target field found ...
        if target_field is None:
            logger.info("Skipping migration for field '%s'" % fieldname)
            return False

        if accessor:
            # get the source field value from the accessor
            value = getattr(self.src, accessor, default)
            if callable(value):
                value = value()
        else:
            # get the source value with the default getter from the field
            value = self.field.get(fieldname)

        # always convert string values to unicode for dexterity fields
        if isinstance(value, six.string_types):
            value = api.safe_unicode(value)

        # set the value with the datamanager (another point to override)
        dm = getMultiAdapter(
            (self.target, target_field), interface=IDataManager)
        if dm:
            dm.set(value)
        else:
            target_field.set(self.target, value)

        logger.info("Migrated field %s -> %s" % (fieldname, target_fieldname))


1			# -- coding: utf-8 --
2
3			import json
4
5			import six
6
7			from Acquisition import aq_parent
8			from bika.lims import api
9			from bika.lims import logger
10			from bika.lims.interfaces import IAuditable
11			from plone.dexterity.interfaces import IDexterityContent
12			from Products.Archetypes.interfaces import IBaseObject
13			from Products.Archetypes.interfaces import IField
14			from senaite.core.interfaces import IContentMigrator
15			from senaite.core.interfaces import IFieldMigrator
16			from senaite.core.migration.utils import copyPermMap
17			from z3c.form.interfaces import IDataManager
18			from zope.component import adapts
19			from zope.component import getMultiAdapter
20			from zope.interface import alsoProvides
21			from zope.interface import directlyProvidedBy
22			from zope.interface import implementer
23
24			SKIP_FIELDS = [
25			"id",
26			"allowDiscussion",
27			"subject",
28			"location",
29			"contributors",
30			"creators",
31			"effectiveDate",
32			"expirationDate",
33			"language",
34			"rights",
35			"creation_date",
36			]
37
38
39			@implementer(IContentMigrator)
40			class ContentMigrator(object):
41			"""SENAITE content migrator
42			"""
43			def __init__(self, src, target):
44			self.src = src
45			self.target = target
46
47			def migrate(self):
48			"""Run the migration
49			"""
50			raise NotImplementedError("Must be implemented by subclass")
51
52			def uncatalog_object(self, obj):
53			"""Uncatalog the object for all catalogs
54			"""
55			# uncatalog from registered catalogs
56			obj.unindexObject()
57			# explicitly uncatalog from uid_catalog
58			uid_catalog = api.get_tool("uid_catalog")
59			url = "/".join(obj.getPhysicalPath()[2:])
60			uid_catalog.uncatalog_object(url)
61
62			def catalog_object(self, obj):
63			"""Catalog the object
64			"""
65			obj.reindexObject()
66
67			def copy_uid(self, obj, uid):
68			"""Set uid on object
69			"""
70			if api.is_dexterity_content(obj):
71			setattr(obj, "_plone.uuid", uid)
72			elif api.is_at_content(obj):
73			setattr(obj, "_at_uid", uid)
74			else:
75			raise TypeError("Cannot set UID on that object")
76
77			def copy_dates(self, src, target):
78			"""copy modification/creation date
79			"""
80			created = api.get_creation_date(src)
81			modified = api.get_modification_date(src)
82			target.creation_date = created
83			target.setModificationDate(modified)
84
85			def copy_creators(self, src, target):
86			"""Copy creators
87			"""
88			target.setCreators(src.listCreators())
89
90			def copy_workflow_history(self, src, target):
91			"""Copy workflow history
92			"""
93			wfh = getattr(src, "workflow_history", None)
94			if wfh:
95			wfh = copyPermMap(wfh)
96			target.workflow_history = wfh
97
98			def copy_marker_interfaces(self, src, target):
99			"""Copy marker interfaces
100			"""
101			alsoProvides(target, directlyProvidedBy(src))
102
103			def copy_snapshots(self, src, target):
104			"""copy over snapshots from source -> target
105			"""
106			snapshots = api.snapshot.get_snapshots(src)
107			storage = api.snapshot.get_storage(target)
108			storage[:] = map(json.dumps, snapshots)[:]
109			alsoProvides(target, IAuditable)
110
111			def delete_object(self, obj):
112			"""delete the object w/o firing events
113			"""
114			self.uncatalog_object(obj)
115			parent = aq_parent(obj)
116			parent._delObject(obj.getId(), suppress_events=True)
117
118			def copy_fields(self, src, target, mapping):
119			"""Copy fields
120			"""
121			src_fields = api.get_fields(src)
122			for fname, field in src_fields.items():
123			if fname in SKIP_FIELDS:
124			continue
125			field_migrator = getMultiAdapter(
126			(field, src, target), interface=IFieldMigrator)
127			# migrate the field
128			field_migrator.migrate(mapping)
129
130
131			class ATDXContentMigrator(ContentMigrator):
132			"""Migrate from AT to DX contents
133			"""
134			adapts(IBaseObject, IDexterityContent)
135
136			def migrate(self, mapping=None, delete_src=True):
137			"""Migrate AT content to DX
138
139			:param mapping: a mapping from source schema field name to a tuple of
140			(accessor name, target field name, default value)
141			"""
142			if mapping is None:
143			mapping = {}
144
145			# copy_fields
146			self.copy_fields(self.src, self.target, mapping)
147
148			# copy the UID
149			self.copy_uid(self.src, self.target)
150
151			# copy auditlog
152			self.copy_snapshots(self.src, self.target)
153
154			# copy creators
155			self.copy_creators(self.src, self.target)
156
157			# copy workflow history
158			self.copy_workflow_history(self.src, self.target)
159
160			# copy marker interfaces
161			self.copy_marker_interfaces(self.src, self.target)
162
163			# copy dates
164			self.copy_dates(self.src, self.target)
165
166			# uncatalog the source object
167			self.uncatalog_object(self.src)
168
169			# reindex the new object
170			self.catalog_object(self.target)
171
172			# delete source object if requested
173			if delete_src:
174			self.delete_object(self.src)
175
176
177			@implementer(IFieldMigrator)
178			class FieldMigrator(object):
179			"""SENAITE field migrator
180			"""
181			def __init__(self, field, src, target):
182			self.field = field
183			self.src = src
184			self.target = target
185
186			def migrate(self):
187			raise NotImplementedError("Must be implemented by subclass")
188
189
190			class ATDXFieldMigrator(FieldMigrator):
191			"""SENAITE AT to DX field migrator
192			"""
193			adapts(IField, IBaseObject, IDexterityContent)
194
195			def migrate(self, mapping):
196			# get all fields on the target
197			target_fields = api.get_fields(self.target)
198
199			fieldname = self.field.getName()
200
201			# check if we have a mapping for this field
202			accessor, target_fieldname, default = (None, None, None, )
203			if fieldname in mapping:
204			accessor, target_fieldname, default = mapping[fieldname]
205
206			target_field = None
207			if target_fieldname:
208			# get the target field with the mapped name
209			target_field = target_fields.get(target_fieldname)
210			else:
211			# check for a field with the same name on the target
212			target_field = target_fields.get(fieldname)
213
214			# no target field found ...
215			if target_field is None:
216			logger.info("Skipping migration for field '%s'" % fieldname)
217			return False
218
219			if accessor:
220			# get the source field value from the accessor
221			value = getattr(self.src, accessor, default)
222			if callable(value):
223			value = value()
224			else:
225			# get the source value with the default getter from the field
226			value = self.field.get(fieldname)
227
228			# always convert string values to unicode for dexterity fields
229			if isinstance(value, six.string_types):
230			value = api.safe_unicode(value)
231
232			# set the value with the datamanager (another point to override)
233			dm = getMultiAdapter(
234			(self.target, target_field), interface=IDataManager)
235			if dm:
236			dm.set(value)
237			else:
238			target_field.set(self.target, value)
239
240			logger.info("Migrated field %s -> %s" % (fieldname, target_fieldname))
241

senaite / senaite.core

Pull Request — 2.x (#1911)

ATDXContentMigrator.migrate() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like