Source code for anyblok_io.bloks.io_xml.importer

# This file is a part of the AnyBlok project
#    Copyright (C) 2015 Jean-Sebastien SUZANNE <>
# This Source Code Form is subject to the terms of the Mozilla Public License,
# v. 2.0. If a copy of the MPL was not distributed with this file,You can
# obtain one at
from anyblok import Declarations
from lxml import etree
from .exceptions import XMLImporterException

register = Declarations.register
IO = Declarations.Model.IO
if_exist = 'overwrite'
if_does_not_exist = 'create'
on_error = 'ignore'

[docs]@register(IO) class Importer: @classmethod def get_mode_choices(cls): res = super(Importer, cls).get_mode_choices() res.update({'Model.IO.Importer.XML': 'XML'}) return res
[docs]@register(IO.Importer) class XML: def __init__(self, importer, blokname=None): self.importer = importer self.error_found = [] self.created_entries = [] self.updated_entries = [] self.params = {} self.two_way_external_id = {} self.blokname = blokname def commit(self): if self.error_found: return False self.importer.commit() return True def _raise(self, msg, on_error=on_error, **kwargs): self.error_found.append(str(msg)) if on_error == 'raise': raise XMLImporterException(msg) def find_entry(self, model=None, external_id=None, param=None, **kwargs): if (model, param) in self.params: return self.params[(model, param)] if model and external_id: if (model, external_id) in self.two_way_external_id: return self.two_way_external_id[(model, external_id)] entry = self.anyblok.IO.Mapping.get(model, external_id) if entry: return entry return None def check_entry_before_import(self, record, entry, if_exist=if_exist, if_does_not_exist=if_does_not_exist, **kwargs): if entry: if if_exist == 'pass': return True, entry elif if_exist == 'raise': self._raise("The tag %r (%r) already exist" % ( record.tag, record.attrib), **kwargs) # case of raise at the end, must not bloc import return True, None else: if if_does_not_exist == 'pass': return True, None elif if_does_not_exist == 'raise': self._raise("The tag %r (%r) does not exist" % ( record.tag, record.attrib), **kwargs) # case of raise at the end, must not bloc import return True, None return False, entry def update_x2M(self, entry, inValues, notInValues): for field in inValues.keys(): if field in notInValues: continue # temporaly deactivate auto flush to let time # orm to fill the remote column with foreign_key # which should be not null with self.anyblok.session.no_autoflush: getattr(entry, field).extend(inValues[field]) def create_entry(self, Model, values, two_way, **kwargs): try: insert_values = {x: y for x, y in values.items() if not isinstance(y, list)} if two_way: return_entry = Model(**insert_values) else: return_entry = Model.insert(**insert_values) self.update_x2M(return_entry, values, insert_values) self.created_entries.append(return_entry) return return_entry except Exception as e: self._raise(e, **kwargs) def map_imported_entry(self, model, param, external_id, two_way, entry, if_exist=if_exist, **kwargs): if entry: if model and param: if (model, param) not in self.params: self.params[(model, param)] = entry else: if self.params[(model, param)] != entry: self._raise("Overwrite the params (%r, %r)" % ( model, param), **kwargs) if external_id: if two_way: self.two_way_external_id[( model, external_id)] = entry else: raiseifexist = if_exist != 'overwrite' self.anyblok.IO.Mapping.set(external_id, entry, blokname=self.blokname, raiseifexist=raiseifexist) def import_entry(self, entry, values, model=None, external_id=None, param=None, if_exist=if_exist, if_does_not_exist=if_does_not_exist, two_way=False, **kwargs): Model = self.anyblok.get(model) return_entry = entry if entry: if if_exist == 'continue': return_entry = entry elif if_exist == 'create': return_entry = self.create_entry( Model, values, two_way, **kwargs) elif if_exist == 'overwrite': try: insert_values = {x: y for x, y in values.items() if not isinstance(y, list)} if insert_values: entry.update(**insert_values) self.update_x2M(entry, values, insert_values) self.updated_entries.append(entry) except Exception as e: self._raise(e, **kwargs) elif if_does_not_exist == 'create': return_entry = self.create_entry(Model, values, two_way, **kwargs) self.map_imported_entry( model, param, external_id, two_way, return_entry, if_exist=if_exist, **kwargs) return return_entry def import_multi_values(self, records, model): vals = [] for record in records: val = self.import_record( record, model=record.attrib.get('model', model), two_way=True ) if not val: continue vals.append(val) return vals def import_value(self, record, ctype, model, on_error=on_error): val = self.import_record(record, model=model) if ctype in ('One2One', 'Many2One'): return val else: if not val: return None pks = val.to_primary_keys() vals = [x for x in pks.values()] if len(vals) > 1: self._raise( "Multi foreign key for %r (%r) are not " "implemented" % (record.tag, record.attrib), on_error=on_error) return vals[0] def get_from_param(self, param, model): if param: if model: if (model, param) in self.params: return self.params[(model, param)] else: self._raise('Param %r waiting for None model ' % param, on_error=on_error) return None def import_field(self, field, ctype, model=None, on_error=on_error): model = field.attrib.get('model', model) param = field.attrib.get('param') res = self.get_from_param(param, model) if res: return res val = field.text external_id = False if 'external_id' in field.attrib: external_id = True val = field.attrib['external_id'] try: res = None if ctype in ('Many2One', 'One2One') and external_id: if model and (model, val) in self.two_way_external_id: res = self.two_way_external_id[(model, val)] if not res: res = self.importer.str2value(val, ctype, external_id=external_id, model=model) if param: self.params[(model, param)] = res return res except Exception as e: self._raise(e, on_error=on_error) return None def import_record(self, record, two_way=False, **kwargs): kwargs.update(record.attrib) _kw = {} if 'on_error' in kwargs: _kw['on_error'] = kwargs['on_error'] entry = self.find_entry(**kwargs) mustbereturned, entry = self.check_entry_before_import( record, entry, **kwargs) if mustbereturned: return entry if 'model' not in kwargs: self._raise("No model found for record %r (%r)" % ( record.tag, kwargs), **_kw) return None Model = self.anyblok.get(kwargs['model']) fields_description = Model.fields_description() values = {} for field in record.getchildren(): if 'on_error' in field.attrib: _kw['on_error'] = field.attrib['on_error'] if not self.validate_field(field, Model, fields_description, **_kw): continue field_name = field.attrib['name'] field_model = fields_description[field_name]['model'] field_type = fields_description[field_name]['type'] val = self._import_record(field, field_model, field_type, **_kw) values[field_name] = val entry = self.import_entry(entry, values, two_way=two_way, **kwargs) if not two_way and self.two_way_external_id: self.two_ways_to_external_id() return entry def two_ways_to_external_id(self): for k, entry in self.two_way_external_id.items(): model, external_id = k # FIXME no controle of if_exist self.map_imported_entry(model, None, external_id, False, entry) def validate_field(self, field, Model, fields_description, **_kw): if field.tag is etree.Comment: return False if field.tag.lower() != 'field': self._raise("Waitting 'field' node, not %r" % field.tag, **_kw) return False if 'name' not in field.attrib: self._raise("field %r (%r) have not attribute name" % ( field.tag, field.attrib), **_kw) return False field_name = field.attrib['name'] if field_name not in fields_description: self._raise('Model %r have not field %r' % ( Model.__registry_name__, field_name)) return False return True def _import_record(self, field, field_model, field_type, **_kw): children = field.getchildren() if children: if field_type in ('One2Many', 'Many2Many'): vals = self.import_multi_values(children, field_model) if vals: return vals return [] else: return self.import_value( field, field_type, field_model, **_kw) else: return self.import_field( field, field_type, model=field_model, **_kw) return None def import_records(self, records): for record in records.getchildren(): if record.tag is etree.Comment: continue elif record.tag.lower() == 'record': self.import_record(record, model=self.importer.model) elif record.tag.lower() == 'commit': self.commit() else: self._raise('%r is not known' % record.tag, **records.attrib) def run(self): records = etree.fromstring(self.importer.file_to_import) if records.tag.lower() == 'records': self.import_records(records) else: self._raise('%r is not known' % records.tag) if self.error_found: if records.attrib.get('on_error', 'raise') == 'raise': msg = "Exception found : \n %s" % '\n'.join(self.error_found) raise XMLImporterException(msg) return { 'error_found': self.error_found, 'created_entries': self.created_entries, 'updated_entries': self.updated_entries, } @classmethod def insert(cls, delimiter=None, quotechar=None, **kwargs): kwargs['mode'] = cls.__registry_name__ if 'model' not in kwargs: raise XMLImporterException( "The column 'model' is required") if not isinstance(kwargs['model'], str): kwargs['model'] = kwargs['model'].__registry_name__ return cls.anyblok.IO.Importer.insert(**kwargs)