niamoto.core.plugins package

Subpackages

Submodules

niamoto.core.plugins.base module

Core base definitions for the Niamoto plugin system.

This module defines the fundamental building blocks for all plugins, including the plugin type enumeration, the abstract base class for plugins, specific abstract base classes for different plugin types (Loaders, Transformers, Exporters, Widgets), and the registration decorator.

class niamoto.core.plugins.base.PluginType(*values)

Bases: Enum

Enumeration of the different types of plugins supported.

TRANSFORMER = 'transformer'
EXPORTER = 'exporter'
WIDGET = 'widget'
LOADER = 'loader'
DEPLOYER = 'deployer'
class niamoto.core.plugins.base.Plugin(db, registry=None)

Bases: ABC

Abstract base class for all plugins in the Niamoto system.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType
resolve_entity_table(entity_name)

Helper method to resolve an entity name to its physical table name.

Parameters:

entity_name (str) – Logical entity name (e.g., ‘taxonomy’, ‘plots’, ‘occurrences’)

Returns:

Physical table name (e.g., ‘entity_taxonomy’, ‘dataset_occurrences’). Returns the original entity_name if the registry is not available (backward compatibility fallback).

Raises:

ValueError – If entity not found in registry

Return type:

str

class niamoto.core.plugins.base.LoaderPlugin(db, registry=None)

Bases: Plugin, ABC

Abstract base class for data loader plugins.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType = 'loader'
abstractmethod load_data(*args, **kwargs)

Load data based on the provided configuration.

The exact signature (args, kwargs) will depend on how the loader is called by the service managing data loading. It likely needs access to its specific configuration parameters.

Returns:

A pandas DataFrame containing the loaded data.

Return type:

DataFrame

class niamoto.core.plugins.base.TransformerPlugin(db, registry=None)

Bases: Plugin, ABC

Abstract base class for data transformer plugins.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType = 'transformer'
output_structure: Dict[str, str] | None = None
abstractmethod transform(data, params)

Transform the input data based on validated parameters.

Parameters:

  • data (Any) – The input data to transform (type might vary, e.g., DataFrame).

  • params (BaseModel) – A Pydantic model instance containing the validated parameters specific to this transformer, derived from its config_model.

Returns:

The transformed data (type might vary).

Return type:

Any

prepare_batch(items, params)

Optionally prepare shared state for a batch of items before repeated transforms.

Exporters can call this once per group to let a transformer preload data or memoize lookups that would otherwise be repeated for every item. The default implementation is a no-op so existing transformers do not need to change.

Parameters:

  • items (List[Any]) – The items that will be transformed next.

  • params (BaseModel) – The validated transformer parameters for the current group.

Return type:

None

class niamoto.core.plugins.base.ExporterPlugin(db, registry=None)

Bases: Plugin, ABC

Abstract base class for data exporter plugins.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType = 'exporter'
abstractmethod export(target_config, repository)

Export data based on the provided target configuration.

Parameters:

  • target_config (TargetConfig) – The validated Pydantic model representing the configuration for this specific export target (from export.yml). Needs forward reference or deferred import.

  • repository (Any) – An object providing access to the transformed data needed for the export.

Return type:

None

class niamoto.core.plugins.base.WidgetPlugin(db, registry=None)

Bases: Plugin, ABC

Abstract base class for visualization widget plugins used in HTML exports.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType = 'widget'
compatible_structures: List[Dict[str, str]] | None = None
get_dependencies()

Declare any external JS or CSS files required by this widget.

These URLs will be collected by the HtmlPageExporter and potentially added to the <head> of the HTML document.

Returns:

A list of strings, typically URLs to CSS or JS files.

Return type:

List[str]

abstractmethod render(data, params)

Render the widget’s HTML representation based on input data and parameters.

Parameters:

  • data (Any) – The specific data required by this widget, extracted based on the ‘data_source’ key in the widget configuration.

  • params (BaseModel) – A Pydantic model instance containing the validated parameters specific to this widget, derived from its config_model.

Returns:

An HTML string representing the rendered widget.

Return type:

str

get_container_html(widget_id, content, config)

Generate the standard HTML container wrapping the widget’s content.

This provides a consistent structure (div, optional title, description) around the HTML generated by the render method.

Parameters:

  • widget_id (str) – A unique ID generated for this specific widget instance on the page.

  • content (str) – The HTML content generated by the widget’s render method.

  • config (WidgetConfig) – The validated Pydantic model (WidgetConfig) representing this widget’s configuration entry from the export.yml file (contains plugin name, data_source, params dict, title, etc.). Needs forward reference or deferred import.

Returns:

The complete HTML string for the widget including its container.

Return type:

str

class niamoto.core.plugins.base.DeployerPlugin(db=None, registry=None)

Bases: Plugin, ABC

Abstract base class for deployment platform plugins.

Deployers publish static sites to hosting platforms (Cloudflare, GitHub Pages, Netlify, etc.) via their HTTP APIs. They don’t need database access.

Parameters:

  • db (Any)

  • registry (Any | None)

type: PluginType = 'deployer'
abstractmethod async deploy(config)

Deploy files and yield SSE log lines.

Parameters:

config (Any) – A DeployConfig instance with platform, exports_dir, project_name, etc.

Return type:

AsyncIterator[str]

Yields strings in the format: - Regular log: “data: some messagenn” - Error: “data: ERROR: messagenn” - Success: “data: SUCCESS: messagenn” - URL: “data: URL: https://…nn” - Done: “data: DONEnn”

async unpublish(config)

Remove a deployed site from the platform and yield SSE log lines.

Default implementation returns an error. Override in subclasses that support unpublishing.

Parameters:

config (Any)

Return type:

AsyncIterator[str]

validate_exports(config)

Pre-flight validation of export directory. Returns list of errors.

Parameters:

config (Any)

Return type:

list[str]

static sse_log(message)

Format a message as an SSE data line.

Parameters:

message (str)

Return type:

str

static sse_error(message)

Format an error as an SSE data line.

Parameters:

message (str)

Return type:

str

static sse_success(message)

Format a success as an SSE data line.

Parameters:

message (str)

Return type:

str

static sse_url(url)

Format a URL as an SSE data line.

Parameters:

url (str)

Return type:

str

static sse_done()

Format the done signal as an SSE data line.

Return type:

str

niamoto.core.plugins.base.register(name, plugin_type=None)

Class decorator to register a plugin with the PluginRegistry.

Parameters:

  • name (str) – The unique identifier (string name) for the plugin.

  • plugin_type (PluginType | None) – The type of the plugin (PluginType Enum). If None, it will be inferred from the ‘type’ class attribute of the decorated class.

Returns:

The original class, after registration.

Raises:

  • TypeError – If the decorated class does not have a ‘type’ attribute and plugin_type is not provided.

  • PluginRegistrationError – If registration fails (e.g., duplicate name for the same type).

niamoto.core.plugins.exceptions module

Custom exceptions for the Niamoto plugin system.

exception niamoto.core.plugins.exceptions.PluginError(message, details=None)

Bases: Exception

Base exception class for all plugin-related errors.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginRegistrationError(message, details=None)

Bases: PluginError

Error raised when plugin registration fails.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginNotFoundError(message, details=None)

Bases: PluginError

Error raised when a requested plugin is not found in the registry.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginConfigError(message, details=None)

Bases: PluginError

Error raised when plugin configuration is invalid.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginLoadError(message, details=None)

Bases: PluginError

Error raised when loading a plugin fails.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginValidationError(message, details=None)

Bases: PluginError

Error raised when plugin validation fails.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginExecutionError(message, details=None)

Bases: PluginError

Error raised when plugin execution fails.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginDependencyError(message, details=None)

Bases: PluginError

Error raised when plugin dependencies are missing or incompatible.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginTypeError(message, details=None)

Bases: PluginError

Error raised when plugin type is invalid or incompatible.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

exception niamoto.core.plugins.exceptions.PluginStateError(message, details=None)

Bases: PluginError

Error raised when plugin is in an invalid state.

Parameters:

  • message (str)

  • details (Dict[str, Any] | None)

niamoto.core.plugins.models module

Pydantic models for validating the Niamoto export.yml configuration file.

This module defines the overall structure of the export configuration, including targets, groups, static pages, widgets, and base configurations for plugins and their parameters. Specific parameter models for individual plugins (e.g., BarPlotParams, InfoGridParams) are defined within the plugin files themselves.

class niamoto.core.plugins.models.BasePluginParams(**extra_data)

Bases: BaseModel

Base model for the ‘params’ dictionary within plugin configurations. Specific plugins should define their own parameter model inheriting from this. Allows extra fields by default if needed, or set model_config = ConfigDict(extra=’forbid’) in subclasses for stricter validation.

Parameters:

extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.PluginConfig(*, plugin, source=None, params=<factory>)

Bases: BaseModel

Base model representing a plugin configuration entry in YAML (e.g., within a list of transformers or potentially exporters/loaders if used).

Parameters:

  • plugin (str)

  • source (str | None)

  • params (Dict[str, Any])

plugin: str
source: str | None
params: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.WidgetLayoutConfig(*, colspan=1, order=0)

Bases: BaseModel

Layout configuration for a widget.

Parameters:

  • colspan (int)

  • order (int)

colspan: int
order: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.WidgetConfig(*, plugin, data_source, title=None, description=None, params=<factory>, layout=None)

Bases: BaseModel

Model representing a widget configuration within the ‘widgets’ list of a ‘web_pages’ target group.

Parameters:

  • plugin (str)

  • data_source (str)

  • title (str | Dict[str, str] | None)

  • description (str | Dict[str, str] | None)

  • params (Dict[str, Any])

  • layout (WidgetLayoutConfig | None)

plugin: str
data_source: str
title: str | Dict[str, str] | None
description: str | Dict[str, str] | None
params: Dict[str, Any]
layout: WidgetLayoutConfig | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.SiteConfig(*, title='Niamoto Data Export', logo_header=None, logo_footer=None, lang='en', languages=None, language_switcher=False, primary_color='#228b22', nav_color='#228b22', **extra_data)

Bases: BaseModel

Global site configuration options.

Parameters:

  • title (str | Dict[str, str])

  • logo_header (str | None)

  • logo_footer (str | None)

  • lang (str)

  • languages (List[str] | None)

  • language_switcher (bool)

  • primary_color (str)

  • nav_color (str)

  • extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

title: str | Dict[str, str]
logo_header: str | None
lang: str
languages: List[str] | None
language_switcher: bool
primary_color: str
nav_color: str
class niamoto.core.plugins.models.NavigationItem(*, text, url=None, children=None)

Bases: BaseModel

A single item in the site navigation menu.

Parameters:

  • text (str | Dict[str, str])

  • url (str | None)

  • children (List[NavigationItem] | None)

text: str | Dict[str, str]
url: str | None
children: List[NavigationItem] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Bases: BaseModel

A single link in a footer section.

Parameters:

  • text (str | Dict[str, str])

  • url (str)

  • external (bool)

text: str | Dict[str, str]
url: str
external: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.FooterSection(*, title, links=<factory>)

Bases: BaseModel

A footer section with a title and a list of links.

Parameters:

  • title (str | Dict[str, str])

  • links (List[FooterLink])

title: str | Dict[str, str]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.StaticPageContext(*, title=None, content_source=None, **extra_data)

Bases: BaseModel

Context data for a static page.

Parameters:

  • title (str | Dict[str, str] | None)

  • content_source (str | None)

  • extra_data (Any)

title: str | Dict[str, str] | None
content_source: str | None
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.StaticPageConfig(*, name, template, output_file, context=<factory>)

Bases: BaseModel

Configuration for a single static page.

Parameters:
name: str
template: str
output_file: str
context: StaticPageContext | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.HtmlExporterParams(*, template_dir, output_dir, base_template='_base.html', copy_assets_from=<factory>, site=<factory>, navigation=<factory>, footer_navigation=<factory>, include_default_assets=True, **extra_data)

Bases: BasePluginParams

Parameters specific to the ‘html_page_exporter’.

Parameters:

  • template_dir (str)

  • output_dir (str)

  • base_template (str)

  • copy_assets_from (List[str])

  • site (SiteConfig)

  • navigation (List[NavigationItem])

  • footer_navigation (List[FooterSection])

  • include_default_assets (bool)

  • extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'json_schema_extra': {'description': 'Generate static HTML website with templates and widgets', 'examples': [{'base_template': '_base.html', 'include_default_assets': True, 'output_dir': 'site', 'template_dir': 'templates'}]}}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

template_dir: str
output_dir: str
base_template: str
copy_assets_from: List[str]
site: SiteConfig
navigation: List[NavigationItem]
footer_navigation: List[FooterSection]
include_default_assets: bool
class niamoto.core.plugins.models.IndexGeneratorDisplayField(*, name, source, fallback=None, type='text', label=None, searchable=False, format=None, mapping=None, filter_options=None, dynamic_options=False, display='normal', inline_badge=False, badge_color=None, badge_style=None, badge_colors=None, badge_styles=None, true_label=None, false_label=None, tooltip_mapping=None, link_template=None, link_label=None, link_title=None, link_target=None, css_class=None, css_style=None, image_fields=None)

Bases: BaseModel

Configuration for a display field in the index generator.

Parameters:

  • name (str)

  • source (str)

  • fallback (str | None)

  • type (str)

  • label (str | Dict[str, str] | None)

  • searchable (bool)

  • format (str | None)

  • mapping (Dict[str, str] | None)

  • filter_options (List[Dict[str, str]] | None)

  • dynamic_options (bool)

  • display (str)

  • inline_badge (bool)

  • badge_color (str | None)

  • badge_style (str | None)

  • badge_colors (Dict[str, str] | None)

  • badge_styles (Dict[str, str] | None)

  • true_label (str | Dict[str, str] | None)

  • false_label (str | Dict[str, str] | None)

  • tooltip_mapping (Dict[str, str] | None)

  • link_template (str | None)

  • link_label (str | Dict[str, str] | None)

  • link_title (str | Dict[str, str] | None)

  • link_target (str | None)

  • css_class (str | None)

  • css_style (str | None)

  • image_fields (Dict[str, str] | None)

name: str
source: str
fallback: str | None
type: str
label: str | Dict[str, str] | None
searchable: bool
format: str | None
mapping: Dict[str, str] | None
filter_options: List[Dict[str, str]] | None
dynamic_options: bool
display: str
inline_badge: bool
badge_color: str | None
badge_style: str | None
badge_colors: Dict[str, str] | None
badge_styles: Dict[str, str] | None
true_label: str | Dict[str, str] | None
false_label: str | Dict[str, str] | None
tooltip_mapping: Dict[str, str] | None
css_class: str | None
css_style: str | None
image_fields: Dict[str, str] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexGeneratorFilterConfig(*, field, values, operator='in')

Bases: BaseModel

Configuration for filtering items in index generator.

Parameters:

  • field (str)

  • values (List[str | int | bool])

  • operator (str)

field: str
values: List[str | int | bool]
operator: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexGeneratorPageConfig(*, title, description=None, items_per_page=20)

Bases: BaseModel

Configuration for the page display in index generator.

Parameters:

  • title (str | Dict[str, str])

  • description (str | Dict[str, str] | None)

  • items_per_page (int)

title: str | Dict[str, str]
description: str | Dict[str, str] | None
items_per_page: int
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexGeneratorViewConfig(*, type, template=None, default=False)

Bases: BaseModel

Configuration for display views in index generator.

Parameters:

  • type (str)

  • template (str | None)

  • default (bool)

type: str
template: str | None
default: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexGeneratorConfig(*, enabled=True, template='_group_index.html', page_config, filters=None, display_fields, views=None, output_pattern='{group_by}/{id}.html')

Bases: BaseModel

Complete configuration for the index generator.

Parameters:
enabled: bool
template: str
page_config: IndexGeneratorPageConfig
filters: List[IndexGeneratorFilterConfig] | None
display_fields: List[IndexGeneratorDisplayField]
views: List[IndexGeneratorViewConfig] | None
output_pattern: str
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.GroupConfigWeb(*, group_by, data_source=None, navigation_entity=None, index_template=None, page_template=None, output_pattern='{group_by}/{id}.html', index_output_pattern='{group_by}/index.html', widgets, index_generator=None)

Bases: BaseModel

Configuration for a group within a ‘web_pages’ target.

Parameters:

  • group_by (str)

  • data_source (str | None)

  • navigation_entity (str | None)

  • index_template (str | None)

  • page_template (str | None)

  • output_pattern (str)

  • index_output_pattern (str)

  • widgets (List[WidgetConfig])

  • index_generator (IndexGeneratorConfig | None)

group_by: str
data_source: str | None
navigation_entity: str | None
index_template: str | None
page_template: str | None
output_pattern: str
index_output_pattern: str
widgets: List[WidgetConfig]
index_generator: IndexGeneratorConfig | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexStructureConfig(*, total_key='total', list_key='{group}', include_total=True)

Bases: BaseModel

Structure configuration for index JSON files.

Parameters:

  • total_key (str)

  • list_key (str)

  • include_total (bool)

total_key: str
list_key: str
include_total: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.JsonOptionsConfig(*, indent=None, ensure_ascii=False)

Bases: BaseModel

JSON serialization options.

Parameters:

  • indent (int | None)

  • ensure_ascii (bool)

indent: int | None
ensure_ascii: bool
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.JsonExporterParams(*, output_dir, detail_output_pattern, index_output_pattern=None, index_structure=<factory>, json_options=<factory>, **extra_data)

Bases: BasePluginParams

Parameters specific to the ‘json_api_exporter’.

Parameters:
output_dir: str
detail_output_pattern: str
index_output_pattern: str | None
index_structure: IndexStructureConfig
json_options: JsonOptionsConfig
model_config: ClassVar[ConfigDict] = {'extra': 'allow'}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.DetailApiConfig(*, pass_through=True, fields=None)

Bases: BaseModel

Configuration for generating detail JSON files.

Parameters:

  • pass_through (bool)

  • fields (List[Any] | None)

pass_through: bool
fields: List[Any] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexApiFieldMapping(*, mapping)

Bases: BaseModel

Represents one field mapping entry in the index configuration.

Parameters:

mapping (Dict[str, Any])

mapping: Dict[str, Any]
classmethod check_single_entry(values)
get_output_key()
Return type:

str

get_config()
Return type:

Any

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.IndexApiConfig(*, fields)

Bases: BaseModel

Configuration for generating index JSON files.

Parameters:

fields (List[IndexApiFieldMapping])

fields: List[IndexApiFieldMapping]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.GroupConfigApi(*, group_by, data_source=None, detail=<factory>, index=None, json_options=None)

Bases: BaseModel

Configuration for a group within a ‘json_api_exporter’ target.

Parameters:
group_by: str
data_source: str | None
detail: DetailApiConfig
index: IndexApiConfig | None
json_options: Dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.DwcMappingValue(*, generator=None, source=None, params=<factory>)

Bases: BaseModel

Represents how to generate a value for a DwC term (generator or source).

Parameters:

  • generator (str | None)

  • source (str | None)

  • params (Dict[str, Any])

generator: str | None
source: str | None
params: Dict[str, Any]
check_generator_or_source()
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.DwcTransformerParams(*, occurrence_list_source, occurrence_table='occurrences', taxonomy_entity='taxonomy', taxon_id_column='id_taxonref', taxon_id_field='id', taxonomy_external_id_column=None, mapping, **extra_data)

Bases: BasePluginParams

Parameters specific to the ‘niamoto_to_dwc_occurrence’ transformer.

Parameters:

  • occurrence_list_source (str)

  • occurrence_table (str)

  • taxonomy_entity (str)

  • taxon_id_column (str)

  • taxon_id_field (str)

  • taxonomy_external_id_column (str | None)

  • mapping (Dict[str, str | DwcMappingValue])

  • extra_data (Any)

model_config: ClassVar[ConfigDict] = {'extra': 'allow', 'json_schema_extra': {'description': 'Transform Niamoto data to Darwin Core Occurrence format', 'examples': [{'mapping': {'decimalLatitude': {'source': 'latitude'}, 'decimalLongitude': {'source': 'longitude'}, 'occurrenceID': {'source': 'occurrence_id'}, 'scientificName': {'source': 'taxon_name'}}, 'occurrence_list_source': 'occurrences', 'occurrence_table': 'dataset_occurrences', 'taxon_id_column': 'id_taxonref', 'taxon_id_field': 'id'}]}}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

occurrence_list_source: str
occurrence_table: str
taxonomy_entity: str
taxon_id_column: str
taxon_id_field: str
taxonomy_external_id_column: str | None
mapping: Dict[str, str | DwcMappingValue]
class niamoto.core.plugins.models.GroupConfigDwc(*, group_by, data_source=None, transformer_plugin, transformer_params, index=None, json_options=None)

Bases: BaseModel

Configuration for a group within a DwC export target.

Parameters:

  • group_by (str)

  • data_source (str | None)

  • transformer_plugin (str)

  • transformer_params (DwcTransformerParams)

  • index (IndexApiConfig | None)

  • json_options (Dict[str, Any] | None)

group_by: str
data_source: str | None
transformer_plugin: str
transformer_params: DwcTransformerParams
index: IndexApiConfig | None
json_options: Dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.TargetConfig(*, name, enabled=True, exporter, params=<factory>, static_pages=None, groups=<factory>)

Bases: BaseModel

Model for a single export target defined in the ‘exports’ list.

Parameters:

  • name (str)

  • enabled (bool)

  • exporter (str)

  • params (Dict[str, Any])

  • static_pages (List[StaticPageConfig] | None)

  • groups (List[Any])

name: str
enabled: bool
exporter: str
params: Dict[str, Any]
static_pages: List[StaticPageConfig] | None
groups: List[Any]
check_exporter_specific_fields()
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class niamoto.core.plugins.models.ExportConfig(*, exports)

Bases: BaseModel

Root model for the entire export.yml configuration file.

Parameters:

exports (List[TargetConfig])

exports: List[TargetConfig]
model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

niamoto.core.plugins.plugin_loader module

Plugin loader module for Niamoto.

This module provides dynamic plugin loading capabilities for both core and project-specific plugins. It handles:

  • Dynamic loading and unloading of plugins

  • Plugin discovery and registration

  • Module path resolution and import management

  • Hot reloading of plugins during development

  • Plugin dependency management

The loader supports a hierarchical plugin structure and maintains plugin state throughout the application lifecycle. It includes safeguards for handling import errors and plugin conflicts.

class niamoto.core.plugins.plugin_loader.PluginInfo(name, plugin_class, scope, path, priority, module_name)

Bases: object

Information about a loaded plugin including its scope and priority

Parameters:

  • name (str)

  • plugin_class (type)

  • scope (str)

  • path (Path)

  • priority (int)

  • module_name (str)

name: str
plugin_class: type
scope: str
path: Path
priority: int
module_name: str
niamoto.core.plugins.plugin_loader.is_plugin_class(obj)

Check if an object is a plugin class.

Parameters:

obj – Object to check

Returns:

True if the object is a plugin class, False otherwise

class niamoto.core.plugins.plugin_loader.PluginLoader

Bases: object

Loader for Niamoto plugins, handling both core and third-party plugins.

Supports cascade resolution across three scopes: - Project-local (~/.niamoto/plugins) - priority 100 - User-global (project/.niamoto/plugins) - priority 50 - System built-in (niamoto/core/plugins) - priority 10

load_plugins_with_cascade(project_path=None)

Load plugins using cascade resolution (Project > User > System).

This is the NEW UNIFIED method that replaces load_core_plugins() + load_project_plugins(). It uses ResourcePaths to discover plugins across all scopes and handles conflicts.

Parameters:

project_path (Path | None) – Optional project path for project-local plugins

Raises:

PluginLoadError – If loading fails

Return type:

None

get_plugin_info()

Get information about loaded plugins.

Returns:

Dictionary containing plugin information

Return type:

Dict[str, Any]

get_plugin_details()

Get detailed information about all loaded plugins including scope and priority.

This is used by the niamoto plugins list command.

Returns:

List of plugin details dictionaries

Return type:

List[Dict[str, Any]]

reload_plugin(module_name)

Reload a specific plugin module.

Parameters:

module_name (str) – Name of module to reload

Raises:

PluginLoadError – If reload fails

Return type:

None

unload_plugin(module_name)

Unload a specific plugin module.

Parameters:

module_name (str) – Name of module to unload

Raises:

PluginLoadError – If unload fails

Return type:

None

discover_plugins(directory)

Discover plugins in a directory without loading them.

Parameters:

directory (Path) – Directory to search for plugins

Returns:

List of discovered plugin information dictionaries

Return type:

List[Dict[str, str]]

register_plugin(module_name, class_name, plugin_type)

Register a plugin without fully loading it.

Parameters:

  • module_name (str) – Name of the module containing the plugin

  • class_name (str) – Name of the plugin class

  • plugin_type (str) – Type of the plugin

Raises:

PluginLoadError – If registration fails

Return type:

None

load_plugin(module_name, class_name)

Load a plugin class from a module.

Parameters:

  • module_name (str) – Name of the module containing the plugin

  • class_name (str) – Name of the plugin class

Returns:

The plugin class

Raises:

PluginLoadError – If loading fails

niamoto.core.plugins.registry module

Plugin registry module for Niamoto.

This module implements a centralized registry system for managing all Niamoto plugins. It provides functionality for:

  • Plugin registration and management

  • Type-safe plugin retrieval

  • Plugin metadata storage and access

  • Plugin type categorization

The registry is implemented as a singleton class that maintains separate registries for different plugin types, ensuring type safety and proper organization of the plugin ecosystem.

class niamoto.core.plugins.registry.PluginRegistry

Bases: object

Central registry for all Niamoto plugins. Handles registration and retrieval of plugins by type.

classmethod register_plugin(name, plugin_class, plugin_type=None, metadata=None)

Register a plugin with the system. :param name: Unique identifier for the plugin :param plugin_class: The plugin class to register :param plugin_type: Type of plugin, inferred from class if not provided :param metadata: Optional metadata about the plugin

Raises:

PluginRegistrationError – If registration fails

Parameters:

  • name (str)

  • plugin_class (Type[Plugin])

  • plugin_type (PluginType | None)

  • metadata (Dict[str, Any] | None)

Return type:

None

classmethod get_plugin(name, plugin_type)

Retrieve a plugin by name and type. :param name: Plugin identifier :param plugin_type: Type of plugin to retrieve

Returns:

The plugin class

Raises:

PluginNotFoundError – If plugin not found

Parameters:
Return type:

Type[Plugin]

classmethod get_plugins_by_type(plugin_type)

Get all plugins of a specific type. :param plugin_type: Type of plugins to retrieve

Returns:

Dictionary of plugin name to plugin class

Parameters:

plugin_type (PluginType)

Return type:

Dict[str, Type[Plugin]]

classmethod get_plugin_metadata(name)

Get metadata for a plugin. :param name: Plugin identifier

Returns:

Plugin metadata or empty dict if none registered

Parameters:

name (str)

Return type:

Dict[str, Any]

classmethod list_plugins()

List all registered plugins by type. :returns: Dictionary of plugin types to list of plugin names

Return type:

Dict[PluginType, list[str]]

classmethod clear()

Clear all registered plugins. Mainly used for testing.

Return type:

None

classmethod has_plugin(name, plugin_type)

Check if a plugin is registered. :param name: Plugin identifier :param plugin_type: Type of plugin to retrieve

Returns:

True if plugin is registered

Parameters:
Return type:

bool

classmethod remove_plugin(name, plugin_type)

Remove a plugin from the registry. :param name: Plugin identifier :param plugin_type: Type of plugin to retrieve

Raises:

PluginNotFoundError – If plugin not found

Parameters:
Return type:

None

Module contents