# -*- coding: utf-8; -*-
################################################################################
#
# WuttaFarm --Web app to integrate with and extend farmOS
# Copyright © 2026 Lance Edgar
#
# This file is part of WuttaFarm.
#
# WuttaFarm 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.
#
# WuttaFarm 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
# WuttaFarm. If not, see <http://www.gnu.org/licenses/>.
#
################################################################################
"""
Custom app handler for WuttaFarm
"""
from wuttjamaican import app as base
[docs]
class WuttaFarmAppHandler(base.AppHandler):
"""
Custom :term:`app handler` for WuttaFarm.
"""
display_format_datetime = "%a, %m/%d/%Y - %H:%M"
default_auth_handler_spec = "wuttafarm.auth:WuttaFarmAuthHandler"
default_install_handler_spec = "wuttafarm.install:WuttaFarmInstallHandler"
[docs]
def get_asset_handler(self):
"""
Get the configured asset handler.
:rtype: :class:`~wuttafarm.assets.AssetHandler`
"""
if "asset" not in self.handlers:
spec = self.config.get(
f"{self.appname}.asset_handler",
default="wuttafarm.assets:AssetHandler",
)
factory = self.load_object(spec)
self.handlers["asset"] = factory(self.config)
return self.handlers["asset"]
[docs]
def get_farmos_handler(self):
"""
Get the configured farmOS integration handler.
:rtype: :class:`~wuttafarm.farmos.FarmOSHandler`
"""
if "farmos" not in self.handlers:
spec = self.config.get(
f"{self.appname}.farmos_handler",
default="wuttafarm.farmos.handler:FarmOSHandler",
)
factory = self.load_object(spec)
self.handlers["farmos"] = factory(self.config)
return self.handlers["farmos"]
[docs]
def get_farmos_integration_mode(self):
"""
Returns the integration mode for farmOS, i.e. to control the
app's behavior regarding that.
"""
enum = self.enum
return self.config.get(
f"{self.appname}.farmos_integration_mode",
default=enum.FARMOS_INTEGRATION_MODE_WRAPPER,
)
[docs]
def is_farmos_mirror(self):
"""
Returns ``True`` if the app is configured in "mirror"
integration mode with regard to farmOS.
"""
enum = self.enum
mode = self.get_farmos_integration_mode()
return mode == enum.FARMOS_INTEGRATION_MODE_MIRROR
[docs]
def is_farmos_wrapper(self):
"""
Returns ``True`` if the app is configured in "wrapper"
integration mode with regard to farmOS.
"""
enum = self.enum
mode = self.get_farmos_integration_mode()
return mode == enum.FARMOS_INTEGRATION_MODE_WRAPPER
[docs]
def is_standalone(self):
"""
Returns ``True`` if the app is configured in "standalone" mode
with regard to farmOS.
"""
enum = self.enum
mode = self.get_farmos_integration_mode()
return mode == enum.FARMOS_INTEGRATION_MODE_NONE
[docs]
def get_farmos_url(self, *args, **kwargs):
"""
Get a farmOS URL. This is a convenience wrapper around
:meth:`~wuttafarm.farmos.handler.FarmOSHandler.get_farmos_url()`.
"""
handler = self.get_farmos_handler()
return handler.get_farmos_url(*args, **kwargs)
[docs]
def get_farmos_client(self, *args, **kwargs):
"""
Get a farmOS client. This is a convenience wrapper around
:meth:`~wuttafarm.farmos.handler.FarmOSHandler.get_farmos_client()`.
"""
handler = self.get_farmos_handler()
return handler.get_farmos_client(*args, **kwargs)
[docs]
def is_farmos_3x(self, *args, **kwargs):
"""
Check if the farmOS version is 3.x. This is a convenience
wrapper around
:meth:`~wuttafarm.farmos.handler.FarmOSHandler.is_farmos_3x()`.
"""
handler = self.get_farmos_handler()
return handler.is_farmos_3x(*args, **kwargs)
[docs]
def is_farmos_4x(self, *args, **kwargs):
"""
Check if the farmOS version is 4.x. This is a convenience
wrapper around
:meth:`~wuttafarm.farmos.handler.FarmOSHandler.is_farmos_4x()`.
"""
handler = self.get_farmos_handler()
return handler.is_farmos_4x(*args, **kwargs)
[docs]
def get_normalizer(self, farmos_client=None):
"""
Get the configured farmOS integration handler.
:rtype: :class:`~wuttafarm.farmos.FarmOSHandler`
"""
spec = self.config.get(
f"{self.appname}.normalizer_spec",
default="wuttafarm.normal:Normalizer",
)
factory = self.load_object(spec)
return factory(self.config, farmos_client)
[docs]
def get_quantity_types(self, session=None):
"""
Returns a list of all known quantity types.
"""
model = self.model
with self.short_session(session=session) as sess:
return (
sess.query(model.QuantityType).order_by(model.QuantityType.name).all()
)
[docs]
def get_measures(self, session=None):
"""
Returns a list of all known measures.
"""
model = self.model
with self.short_session(session=session) as sess:
return sess.query(model.Measure).order_by(model.Measure.ordinal).all()
[docs]
def get_units(self, session=None):
"""
Returns a list of all known units.
"""
model = self.model
with self.short_session(session=session) as sess:
return sess.query(model.Unit).order_by(model.Unit.name).all()
[docs]
def get_material_types(self, session=None):
"""
Returns a list of all known material types.
"""
model = self.model
with self.short_session(session=session) as sess:
return (
sess.query(model.MaterialType).order_by(model.MaterialType.name).all()
)
def get_quantity_models(self):
model = self.model
return {
"standard": model.StandardQuantity,
"material": model.MaterialQuantity,
}
def get_true_quantity(self, quantity, require=True):
model = self.model
if not isinstance(quantity, model.Quantity):
if require and not quantity:
raise ValueError(f"quantity is not valid: {quantity}")
return quantity
session = self.get_session(quantity)
models = self.get_quantity_models()
if require and quantity.quantity_type_id not in models:
raise ValueError(
f"quantity has invalid quantity_type_id: {quantity.quantity_type_id}"
)
true_quantity = session.get(models[quantity.quantity_type_id], quantity.uuid)
if require and not true_quantity:
raise ValueError(f"quantity has no true/typed quantity record: {quantity}")
return true_quantity
def make_true_quantity(self, quantity_type_id, **kwargs):
models = self.get_quantity_models()
kwargs["quantity_type_id"] = quantity_type_id
return models[quantity_type_id](**kwargs)
[docs]
def auto_sync_to_farmos(self, obj, model_name=None, client=None, require=True):
"""
Export the given object to farmOS, using configured handler.
This should ensure the given object is also *updated* with the
farmOS UUID and Drupal ID, when new record is created in
farmOS.
:param obj: Any data object in WuttaFarm, e.g. AnimalAsset
instance.
:param client: Existing farmOS API client to use. If not
specified, a new one will be instantiated.
:param require: If true, this will *require* the export
handler to support objects of the given type. If false,
then nothing will happen / export is silently skipped when
there is no such exporter.
"""
handler = self.app.get_import_handler("export.to_farmos.from_wuttafarm")
if not model_name:
model_name = type(obj).__name__
if model_name not in handler.importers:
if require:
raise ValueError(f"no exporter found for {model_name}")
return
# nb. begin txn to establish the API client
handler.begin_target_transaction(client)
importer = handler.get_importer(model_name, caches_target=False)
normal = importer.normalize_source_object(obj)
importer.process_data(source_data=[normal])
[docs]
def auto_sync_from_farmos(self, obj, model_name, client=None, require=True):
"""
Import the given object from farmOS, using configured handler.
:param obj: Any data record from farmOS.
:param model_name': Model name for the importer to use,
e.g. ``"AnimalAsset"``.
:param client: Existing farmOS API client to use. If not
specified, a new one will be instantiated.
:param require: If true, this will *require* the import
handler to support objects of the given type. If false,
then nothing will happen / import is silently skipped when
there is no such importer.
"""
model = self.app.model
handler = self.app.get_import_handler("import.to_wuttafarm.from_farmos")
if model_name not in handler.importers:
if require:
raise ValueError(f"no importer found for {model_name}")
return
# nb. begin txn to establish the API client
handler.begin_source_transaction(client)
with self.short_session(commit=True) as session:
if user := session.query(model.User).filter_by(username="farmos").first():
session.info["continuum_user_id"] = user.uuid
handler.target_session = session
importer = handler.get_importer(model_name, caches_target=False)
normal = importer.normalize_source_object(obj)
importer.process_data(source_data=[normal])
[docs]
class WuttaFarmAppProvider(base.AppProvider):
"""
The :term:`app provider` for WuttaFarm.
"""
email_modules = ["wuttafarm.emails"]