# -*- coding: utf-8; -*-
################################################################################
#
# WuttaSync -- Wutta Framework for data import/export and real-time sync
# Copyright © 2024-2026 Lance Edgar
#
# This file is part of Wutta Framework.
#
# Wutta Framework is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option) any
# later version.
#
# Wutta Framework 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 General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# Wutta Framework. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Data Import / Export Handlers
"""
# pylint: disable=too-many-lines
import logging
import os
import sys
from collections import OrderedDict
from enum import Enum
import humanize
from wuttjamaican.app import GenericHandler
from wuttjamaican.diffs import Diff
log = logging.getLogger(__name__)
[docs]
class Orientation(Enum):
"""
Enum values for :attr:`ImportHandler.orientation`.
"""
IMPORT = "import"
EXPORT = "export"
[docs]
class ImportHandler( # pylint: disable=too-many-public-methods,too-many-instance-attributes
GenericHandler
):
"""
Base class for all import/export handlers.
Despite the name ``ImportHandler`` this can be used for export as
well. The logic is no different on a technical level and the
"export" concept is mostly only helpful to the user. The latter
is important of course and to help with that we track the
:attr:`orientation` to distinguish.
The role of the "import/export handler" (instance of this class)
is to orchestrate the overall DB connections, transactions and
then invoke the importer/exporter instance(s) to do the actual
data assessment/transfer. Each of the latter will be an instance
of (a subclass of) :class:`~wuttasync.importing.base.Importer`.
"""
source_key = None
"""
Key identifier for the data source.
This should "uniquely" identify the data source, within the
context of the data target. For instance in the case of CSV →
Wutta, ``csv`` is the source key.
Among other things, this value is used in :meth:`get_key()`.
"""
target_key = None
"""
Key identifier for the data target.
This should "uniquely" identify the data target. For instance in
the case of CSV → Wutta, ``wutta`` is the target key.
Among other things, this value is used in :meth:`get_key()`.
"""
orientation = Orientation.IMPORT
"""
Orientation for the data flow. Must be a value from
:enum:`Orientation`:
* ``Orientation.IMPORT`` (aka. ``'import'``)
* ``Orientation.EXPORT`` (aka. ``'export'``)
Note that the value may be displayed to the user where helpful::
print(handler.orientation.value)
See also :attr:`actioning`.
It's important to understand the difference between import/export
and source/target; they are independent concepts. Source and
target indicate where data comes from and where it's going,
whereas import vs. export is mostly cosmetic.
How a given data flow's orientation is determined, is basically up
to the developer. Most of the time it is straightforward,
e.g. CSV → Wutta would be import, and Wutta → CSV would be
export. But confusing edge cases certainly exist, you'll know
them when you see them. In those cases the developer should try
to choose whichever the end user is likely to find less confusing.
"""
dry_run = False
"""
Flag indicating whether data import/export should truly happen vs.
dry-run only.
If true, the data transaction will be rolled back at the end; if
false then it will be committed.
See also :meth:`rollback_transaction()` and
:meth:`commit_transaction()`.
"""
process_started = None
warnings = False
"""
Boolean indicating the import/export should run in "warnings"
mode.
If set, this declares that no changes are expected for the
import/export job. If any changes do occur with this flag set, a
diff warning email is sent within :meth:`process_changes()`.
See also :attr:`warnings_recipients`,
:attr:`warnings_max_diffs` and :attr:`warnings_email_key`.
"""
warnings_email_key = None
"""
Explicit :term:`email key` for sending the diff warning email,
*unique to this import/export type*.
Handlers do not normally set this, so the email key is determined
automatically within :meth:`get_warnings_email_key()`.
See also :attr:`warnings`.
"""
warnings_recipients = None
"""
Explicit recipient list for the warning email. If not set, the
recipients are determined automatically via config.
See also :attr:`warnings`.
"""
warnings_max_diffs = 15
"""
Max number of record diffs (per model) to show in the warning email.
See also :attr:`warnings`.
"""
runas_username = None
"""
Username responsible for running the import/export job. This is
mostly used for Continuum versioning.
"""
transaction_comment = None
"""
Optional comment to apply to the transaction, where applicable.
This is mostly used for Continuum versioning.
"""
importers = None
"""
This should be a dict of all importer/exporter classes available
to the handler. Keys are "model names" and each value is an
importer/exporter class. For instance::
{
'Widget': WidgetImporter,
}
This dict is defined during the handler constructor; see also
:meth:`define_importers()`.
Note that in practice, this is usually an ``OrderedDict`` so that
the "sorting" of importer/exporters can be curated.
If you want an importer/exporter instance you should not use this
directly but instead call :meth:`get_importer()`.
"""
def __init__(self, config, **kwargs):
""" """
super().__init__(config)
# callers can set any attrs they want
for k, v in kwargs.items():
setattr(self, k, v)
self.importers = self.define_importers()
def __str__(self):
""" """
return self.get_title()
@property
def actioner(self):
"""
Convenience property which effectively returns the
:attr:`orientation` as a noun - i.e. one of:
* ``'importer'``
* ``'exporter'``
See also :attr:`actioning`.
"""
return f"{self.orientation.value}er"
@property
def actioning(self):
"""
Convenience property which effectively returns the
:attr:`orientation` in progressive verb tense - i.e. one of:
* ``'importing'``
* ``'exporting'``
See also :attr:`actioner`.
"""
return f"{self.orientation.value}ing"
[docs]
@classmethod
def get_key(cls):
"""
Returns the :term:`import/export key` for the handler. This
is a combination of :attr:`source_key` and :attr:`target_key`
and :attr:`orientation`.
For instance in the case of Wutta → CSV export, the key is:
``export.to_csv.from_wutta``
Note that more than one handler may use the same key; but only
one will be configured as the "designated" handler for that
key, a la
:meth:`~wuttasync.app.WuttaSyncAppProvider.get_import_handler()`.
See also :meth:`get_spec()`.
"""
return f"{cls.orientation.value}.to_{cls.target_key}.from_{cls.source_key}"
[docs]
@classmethod
def get_spec(cls):
"""
Returns the "class spec" for the handler. This value is the
same as what might be used to configure the default handler
for a given key.
For instance in the case of CSV → Wutta, the default handler
spec is ``wuttasync.importing.csv:FromCsvToWutta``.
See also :meth:`get_key()`.
"""
return f"{cls.__module__}:{cls.__name__}"
[docs]
def get_title(self):
"""
Returns the full display title for the handler, e.g. ``"CSV →
Wutta"``.
Note that the :attr:`orientation` is not included in this title.
It calls :meth:`get_source_title()` and
:meth:`get_target_title()` to construct the full title.
"""
source = self.get_source_title()
target = self.get_target_title()
return f"{source} → {target}"
[docs]
def get_source_title(self):
"""
Returns the display title for the data source.
By default this returns :attr:`source_key`, but this can be
overriden by class attribute.
Base class can define ``generic_source_title`` to provide a
new default::
class FromExcelHandler(ImportHandler):
generic_source_title = "Excel File"
Subclass can define ``source_title`` to be explicit::
class FromExcelToWutta(FromExcelHandler, ToWuttaHandler):
source_title = "My Spreadsheet"
See also :meth:`get_title()` and :meth:`get_target_title()`.
"""
if hasattr(self, "source_title"):
return self.source_title
if hasattr(self, "generic_source_title"):
return self.generic_source_title
return self.source_key
[docs]
def get_target_title(self):
"""
Returns the display title for the data target.
By default this returns :attr:`target_key`, but this can be
overriden by class attribute.
Base class can define ``generic_target_title`` to provide a
new default::
class ToExcelHandler(ImportHandler):
generic_target_title = "Excel File"
Subclass can define ``target_title`` to be explicit::
class FromWuttaToExcel(FromWuttaHandler, ToExcelHandler):
target_title = "My Spreadsheet"
See also :meth:`get_title()` and :meth:`get_source_title()`.
"""
if hasattr(self, "target_title"):
return self.target_title
if hasattr(self, "generic_target_title"):
return self.generic_target_title
return self.target_key
[docs]
def process_data(self, *keys, **kwargs):
"""
Run import/export operations for the specified models.
:param \\*keys: One or more importer/exporter (model) keys, as
defined by the handler.
Each key specified must be present in :attr:`importers` and
thus will correspond to an importer/exporter class.
A transaction is begun on the source and/or target side as
needed, then for each model key requested, the corresponding
importer/exporter is created and invoked. And finally the
transaction is committed (assuming normal operation).
See also these methods which may be called from this one:
* :meth:`consume_kwargs()`
* :meth:`begin_transaction()`
* :meth:`get_importer()`
* :meth:`~wuttasync.importing.base.Importer.process_data()` (on the importer/exporter)
* :meth:`process_changes()`
* :meth:`rollback_transaction()`
* :meth:`commit_transaction()`
"""
kwargs = self.consume_kwargs(kwargs)
self.process_started = self.app.localtime()
self.begin_transaction()
changes = OrderedDict()
if not keys:
keys = self.get_default_importer_keys()
success = False
try:
# loop thru specified importer keys
for key in keys:
# invoke importer
importer = self.get_importer(key, **kwargs)
created, updated, deleted = importer.process_data()
changed = bool(created or updated or deleted)
# log what happened
msg = "%s: added %d; updated %d; deleted %d %s records"
if self.dry_run:
msg += " (dry run)"
logger = log.warning if changed and self.warnings else log.info
logger(
msg, self.get_title(), len(created), len(updated), len(deleted), key
)
# keep track of any changes
if changed:
changes[key] = created, updated, deleted
# post-processing for all changes
if changes:
self.process_changes(changes)
success = True
except:
log.exception("what should happen here?") # TODO
raise
finally:
if not success:
log.warning("something failed, so transaction was rolled back")
self.rollback_transaction()
elif self.dry_run:
log.info("dry run, so transaction was rolled back")
self.rollback_transaction()
else:
log.info("transaction was committed")
self.commit_transaction()
[docs]
def consume_kwargs(self, kwargs):
"""
This method is called by :meth:`process_data()`.
Its purpose is to give handlers a hook by which they can
update internal handler state from the given kwargs, prior to
running the import/export task(s).
Any kwargs which pertain only to the handler, should be
removed before they are returned. But any kwargs which (also)
may pertain to the importer/exporter instance, should *not* be
removed, so they are passed along via :meth:`get_importer()`.
:param kwargs: Dict of kwargs, "pre-consumption." This is the
same kwargs dict originally received by
:meth:`process_data()`.
:returns: Dict of kwargs, "post-consumption."
"""
if "dry_run" in kwargs:
self.dry_run = kwargs["dry_run"]
if "warnings" in kwargs:
self.warnings = kwargs.pop("warnings")
if "warnings_recipients" in kwargs:
self.warnings_recipients = self.config.parse_list(
kwargs.pop("warnings_recipients")
)
if "warnings_max_diffs" in kwargs:
self.warnings_max_diffs = kwargs.pop("warnings_max_diffs")
if "runas_username" in kwargs:
self.runas_username = kwargs.pop("runas_username")
if "transaction_comment" in kwargs:
self.transaction_comment = kwargs.pop("transaction_comment")
return kwargs
[docs]
def begin_transaction(self):
"""
Begin an import/export transaction, on source and/or target
side as needed.
This is normally called from :meth:`process_data()`.
Default logic will call both:
* :meth:`begin_source_transaction()`
* :meth:`begin_target_transaction()`
"""
self.begin_source_transaction()
self.begin_target_transaction()
[docs]
def begin_source_transaction(self):
"""
Begin a transaction on the source side, if applicable.
This is normally called from :meth:`begin_transaction()`.
"""
[docs]
def begin_target_transaction(self):
"""
Begin a transaction on the target side, if applicable.
This is normally called from :meth:`begin_transaction()`.
"""
[docs]
def commit_transaction(self):
"""
Commit the current import/export transaction, on source and/or
target side as needed.
This is normally called from :meth:`process_data()`.
Default logic will call both:
* :meth:`commit_target_transaction()`
* :meth:`commit_source_transaction()`
.. note::
By default the target transaction is committed first; this
is to avoid edge case errors when the source connection
times out. In such cases we want to properly cleanup the
target and then if an error happens when trying to cleanup
the source, it is less disruptive.
"""
# nb. it can sometimes be important to commit the target
# transaction first. in particular when the import takes a
# long time, it may be that no activity occurs on the source
# DB session after the initial data read. then at the end
# committing the source transaction may trigger a connection
# timeout error, which then prevents target transaction from
# committing. so now we just commit target first instead.
# TODO: maybe sequence should be configurable?
self.commit_target_transaction()
self.commit_source_transaction()
[docs]
def commit_source_transaction(self):
"""
Commit the transaction on the source side, if applicable.
This is normally called from :meth:`commit_transaction()`.
"""
[docs]
def commit_target_transaction(self):
"""
Commit the transaction on the target side, if applicable.
This is normally called from :meth:`commit_transaction()`.
"""
[docs]
def rollback_transaction(self):
"""
Rollback the current import/export transaction, on source
and/or target side as needed.
This is normally called from :meth:`process_data()`. It is
"always" called when :attr:`dry_run` is true, but also may be
called if errors are encountered.
Default logic will call both:
* :meth:`rollback_target_transaction()`
* :meth:`rollback_source_transaction()`
.. note::
By default the target transaction is rolled back first;
this is to avoid edge case errors when the source
connection times out. In such cases we want to properly
cleanup the target and then if an error happens when trying
to cleanup the source, it is less disruptive.
"""
# nb. it can sometimes be important to rollback the target
# transaction first. in particular when the import takes a
# long time, it may be that no activity occurs on the source
# DB session after the initial data read. then at the end
# rolling back the source transaction may trigger a connection
# timeout error, which then prevents target transaction from
# being rolled back. so now we always rollback target first.
# TODO: maybe sequence should be configurable?
self.rollback_target_transaction()
self.rollback_source_transaction()
[docs]
def rollback_source_transaction(self):
"""
Rollback the transaction on the source side, if applicable.
This is normally called from :meth:`rollback_transaction()`.
"""
[docs]
def rollback_target_transaction(self):
"""
Rollback the transaction on the target side, if applicable.
This is normally called from :meth:`rollback_transaction()`.
"""
[docs]
def define_importers(self):
"""
This method must "define" all importer/exporter classes
available to the handler. It is called from the constructor.
This should return a dict keyed by "model name" and each value
is an importer/exporter class. The end result is then
assigned as :attr:`importers` (in the constoructor).
For instance::
return {
'Widget': WidgetImporter,
}
Note that the model name will be displayed in various places
and the caller may invoke a specific importer/exporter by this
name etc. See also :meth:`get_importer()`.
"""
return OrderedDict()
[docs]
def get_importer(self, key, **kwargs):
"""
Returns an importer/exporter instance corresponding to the
given key.
Note that this will always create a *new* instance; they are
not cached.
The key will be the "model name" mapped to a particular
importer/exporter class and thus must be present in
:attr:`importers`.
This method is called from :meth:`process_data()` but may also
be used by ad-hoc callers elsewhere.
It will call :meth:`get_importer_kwargs()` and then construct
the importer/exporter instance using those kwargs.
:param key: Model key for desired importer/exporter.
:param \\**kwargs: Extra/override kwargs for the importer.
:returns: Instance of (subclass of)
:class:`~wuttasync.importing.base.Importer`.
"""
if key not in self.importers:
orientation = self.orientation.value
raise KeyError(f"unknown {orientation} key: {key}")
kwargs = self.get_importer_kwargs(key, **kwargs)
kwargs["handler"] = self
# nb. default logic should (normally) determine keys
if "keys" in kwargs and not kwargs["keys"]:
del kwargs["keys"]
factory = self.importers[key]
return factory(self.config, **kwargs)
[docs]
def get_importer_kwargs(self, key, **kwargs): # pylint: disable=unused-argument
"""
Returns a dict of kwargs to be used when construcing an
importer/exporter with the given key. This is normally called
from :meth:`get_importer()`.
:param key: Model key for the desired importer/exporter,
e.g. ``'Widget'``
:param \\**kwargs: Any kwargs we have so collected far.
:returns: Final kwargs dict for new importer/exporter.
"""
return kwargs
[docs]
def is_default(self, key): # pylint: disable=unused-argument
"""
Return a boolean indicating whether the importer corresponding
to ``key`` should be considered "default" - i.e. included as
part of a typical "import all" job.
The default logic here returns ``True`` in all cases; subclass can
override as needed.
:param key: Key indicating the importer.
:rtype: bool
"""
return True
[docs]
def get_default_importer_keys(self):
"""
Return the list of importer keys which should be considered
"default" - i.e. which should be included as part of a typical
"import all" job.
This inspects :attr:`importers` and calls :meth:`is_default()`
for each, to determine the result.
:returns: List of importer keys (strings).
"""
keys = list(self.importers)
keys = [k for k in keys if self.is_default(k)]
return keys
[docs]
def process_changes(self, changes):
"""
Run post-processing operations on the given changes, if
applicable.
This method is called by :meth:`process_data()`, if any
changes were made.
Default logic will send a "diff warning" email to the
configured recipient(s), if :attr:`warnings` mode is enabled.
If it is not enabled, nothing happens.
:param changes: :class:`~python:collections.OrderedDict` of
changes from the overall import/export job. The structure
is described below.
Keys for the ``changes`` dict will be model/importer names,
for instance::
{
"Sprocket": {...},
"User": {...},
}
Value for each model key is a 3-tuple of ``(created, updated,
deleted)``. Each of those elements is a list::
{
"Sprocket": (
[...], # created
[...], # updated
[...], # deleted
),
}
The list elements are always tuples, but the structure
varies::
{
"Sprocket": (
[ # created, 2-tuples
(obj, source_data),
],
[ # updated, 3-tuples
(obj, source_data, target_data),
],
[ # deleted, 2-tuples
(obj, target_data),
],
),
}
"""
if not self.warnings:
return
def make_diff(*args, **kwargs):
return Diff(self.config, *args, **kwargs)
runtime = self.app.localtime() - self.process_started
data = {
"handler": self,
"title": self.get_title(),
"source_title": self.get_source_title(),
"target_title": self.get_target_title(),
"dry_run": self.dry_run,
"argv": sys.argv,
"runtime": runtime,
"runtime_display": humanize.naturaldelta(runtime),
"changes": changes,
"make_diff": make_diff,
"max_diffs": self.warnings_max_diffs,
}
# maybe override recipients
kw = {}
if self.warnings_recipients:
kw["to"] = self.warnings_recipients
# TODO: should we in fact clear these..?
kw["cc"] = []
kw["bcc"] = []
# send the email
email_key = self.get_warnings_email_key()
self.app.send_email(email_key, data, fallback_key="import_export_warning", **kw)
log.info("%s: warning email was sent", self.get_title())
[docs]
def get_warnings_email_key(self):
"""
Returns the :term:`email key` to be used for sending the diff
warning email.
The email key should be unique to this import/export type
(really, the :term:`import/export key`) but not necessarily
unique to one handler.
If :attr:`warnings_email_key` is set, it will be used as-is.
Otherwise one is generated from :meth:`get_key()`.
:returns: Email key for diff warnings
"""
if self.warnings_email_key:
return self.warnings_email_key
return self.get_key().replace(".", "_") + "_warning"
[docs]
class FromFileHandler(ImportHandler):
"""
Handler for import/export which uses input file(s) as data source.
This handler assumes its importer/exporter classes inherit from
:class:`~wuttasync.importing.base.FromFile` for source parent
logic.
"""
def process_data(self, *keys, **kwargs): # pylint: disable=empty-docstring
""" """
# interpret file vs. folder path
# nb. this assumes FromFile importer/exporter
path = kwargs.pop("input_file_path", None)
if path:
if not kwargs.get("input_file_dir") and os.path.isdir(path):
kwargs["input_file_dir"] = path
else:
kwargs["input_file_path"] = path
# and carry on
super().process_data(*keys, **kwargs)
[docs]
class FromSqlalchemyHandler(ImportHandler):
"""
Base class for import/export handlers using SQLAlchemy ORM (DB) as
data source.
This is meant to be used with importers/exporters which inherit
from :class:`~wuttasync.importing.base.FromSqlalchemy`. It will
set the
:attr:`~wuttasync.importing.base.FromSqlalchemy.source_session`
attribute when making them; cf. :meth:`get_importer_kwargs()`.
This is the base class for :class:`FromWuttaHandler`, but can be
used with any database.
See also :class:`ToSqlalchemyHandler`.
"""
source_session = None
"""
Reference to the :term:`db session` for data source.
This will be ``None`` unless a transaction is running.
"""
[docs]
def begin_source_transaction(self):
"""
This calls :meth:`make_source_session()` and assigns the
result to :attr:`source_session`.
"""
self.source_session = self.make_source_session()
[docs]
def commit_source_transaction(self):
"""
This commits and closes :attr:`source_session`.
"""
self.source_session.commit()
self.source_session.close()
self.source_session = None
[docs]
def rollback_source_transaction(self):
"""
This rolls back, then closes :attr:`source_session`.
"""
self.source_session.rollback()
self.source_session.close()
self.source_session = None
[docs]
def make_source_session(self):
"""
Make and return a new :term:`db session` for the data source.
Default logic is not implemented; subclass must override.
:returns: :class:`~sqlalchemy.orm.Session` instance
"""
raise NotImplementedError
[docs]
def get_importer_kwargs(self, key, **kwargs):
"""
This modifies the new importer kwargs to add:
* ``source_session`` - reference to :attr:`source_session`
See also docs for parent method,
:meth:`~ImportHandler.get_importer_kwargs()`.
"""
kwargs = super().get_importer_kwargs(key, **kwargs)
kwargs["source_session"] = self.source_session
return kwargs
[docs]
class FromWuttaHandler(FromSqlalchemyHandler):
"""
Handler for import/export which uses Wutta ORM (:term:`app
database`) as data source.
This inherits from :class:`FromSqlalchemyHandler`.
See also :class:`ToWuttaHandler`.
"""
source_key = "wutta"
"" # nb. suppress docs
[docs]
def get_source_title(self):
"""
This overrides default logic to use
:meth:`~wuttjamaican:wuttjamaican.app.AppHandler.get_title()`
as the default value.
Subclass can still define
:attr:`~wuttasync.importing.handlers.ImportHandler.source_title`
(or
:attr:`~wuttasync.importing.handlers.ImportHandler.generic_source_title`)
to customize.
See also docs for parent method:
:meth:`~wuttasync.importing.handlers.ImportHandler.get_source_title()`
"""
if hasattr(self, "source_title"):
return self.source_title
if hasattr(self, "generic_source_title"):
return self.generic_source_title
return self.app.get_title()
[docs]
def make_source_session(self):
"""
This calls
:meth:`~wuttjamaican:wuttjamaican.app.AppHandler.make_session()`
and returns it.
"""
return self.app.make_session()
[docs]
class ToSqlalchemyHandler(ImportHandler):
"""
Base class for import/export handlers which target a SQLAlchemy
ORM (DB).
This is the base class for :class:`ToWuttaHandler`, but can be
used with any database.
See also :class:`FromSqlalchemyHandler`.
"""
target_session = None
"""
Reference to the SQLAlchemy :term:`db session` for the target side.
This may often be a session for the :term:`app database` (i.e. for
importing to Wutta DB) but it could also be any other.
This will be ``None`` unless an import/export transaction is
underway. See also :meth:`begin_target_transaction()`.
"""
[docs]
def begin_target_transaction(self):
"""
Establish a new :term:`db session` via
:meth:`make_target_session()` and assign the result to
:attr:`target_session`.
"""
self.target_session = self.make_target_session()
[docs]
def rollback_target_transaction(self):
"""
Rollback the :attr:`target_session`.
"""
self.target_session.rollback()
self.target_session.close()
self.target_session = None
[docs]
def commit_target_transaction(self):
"""
Commit the :attr:`target_session`.
"""
self.target_session.commit()
self.target_session.close()
self.target_session = None
[docs]
def make_target_session(self):
"""
Make and return a new :term:`db session` for the import/export.
Subclass must override this; default logic is not implemented.
"""
raise NotImplementedError
def get_importer_kwargs(self, key, **kwargs): # pylint: disable=empty-docstring
""" """
kwargs = super().get_importer_kwargs(key, **kwargs)
kwargs.setdefault("target_session", self.target_session)
return kwargs
[docs]
class ToWuttaHandler(ToSqlalchemyHandler):
"""
Handler for import/export which targets Wutta ORM (:term:`app
database`).
This inherits from :class:`ToSqlalchemyHandler`.
See also :class:`FromWuttaHandler`.
"""
target_key = "wutta"
"" # nb. suppress docs
def get_target_title(self): # pylint: disable=empty-docstring
""" """
# nb. we override parent to use app title as default
if hasattr(self, "target_title"):
return self.target_title
if hasattr(self, "generic_target_title"):
return self.generic_target_title
return self.app.get_title()
[docs]
def make_target_session(self):
"""
This creates a typical :term:`db session` for the app by
calling
:meth:`~wuttjamaican:wuttjamaican.app.AppHandler.make_session()`.
It then may "customize" the session slightly. These
customizations only are relevant if Wutta-Continuum versioning
is enabled:
If :attr:`~ImportHandler.runas_username` is set, the
responsible user (``continuum_user_id``) will be set for the
new session as well.
Similarly, if :attr:`~ImportHandler.transaction_comment` is
set, it (``continuum_comment``) will also be set for the new
session.
:returns: :class:`~wuttjamaican:wuttjamaican.db.sess.Session`
instance.
"""
model = self.app.model
session = self.app.make_session()
# set runas user in case continuum versioning is enabled
if self.runas_username:
if user := (
session.query(model.User)
.filter_by(username=self.runas_username)
.first()
):
session.info["continuum_user_id"] = user.uuid
else:
log.warning("runas username not found: %s", self.runas_username)
# set comment in case continuum versioning is enabled
if self.transaction_comment:
session.info["continuum_comment"] = self.transaction_comment
return session