niamoto.gui.api.routers package

Submodules

niamoto.gui.api.routers.bootstrap module

niamoto.gui.api.routers.config module

Configuration management API endpoints for reading and updating YAML configs.

class niamoto.gui.api.routers.config.ConfigUpdate(*, content, backup=True)

Bases: BaseModel

Request body for updating configuration.

Parameters:

  • content (Dict[str, Any] | List[Any])

  • backup (bool)

content: Dict[str, Any] | List[Any]
backup: bool
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ConfigResponse(*, success, message, content=None, backup_path=None)

Bases: BaseModel

Response model for configuration operations.

Parameters:

  • success (bool)

  • message (str)

  • content (Dict[str, Any] | List[Any] | None)

  • backup_path (str | None)

success: bool
message: str
content: Dict[str, Any] | List[Any] | None
backup_path: str | None
model_config: ClassVar[ConfigDict] = {}

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

niamoto.gui.api.routers.config.ensure_config_dir()

Ensure the config directory exists.

niamoto.gui.api.routers.config.create_backup(config_path)

Create a backup of the configuration file.

Parameters:

config_path (Path)

Return type:

Path | None

async niamoto.gui.api.routers.config.get_project_info()

Get project information from config.yml.

Returns:

Project information including name, version, etc.

Return type:

Dict[str, Any]

class niamoto.gui.api.routers.config.HierarchyFields(*, has_nested_set=False, has_parent=False, has_level=False, lft_field=None, rght_field=None, parent_id_field=None, level_field=None, id_field=None, name_field=None)

Bases: BaseModel

Detected hierarchy fields in a reference table.

Parameters:

  • has_nested_set (bool)

  • has_parent (bool)

  • has_level (bool)

  • lft_field (str | None)

  • rght_field (str | None)

  • parent_id_field (str | None)

  • level_field (str | None)

  • id_field (str | None)

  • name_field (str | None)

has_nested_set: bool
has_parent: bool
has_level: bool
lft_field: str | None
rght_field: str | None
parent_id_field: str | None
level_field: str | None
id_field: str | None
name_field: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ReferenceInfo(*, name, table_name, kind, description=None, schema_fields=[], entity_count=None, can_enrich=False, enrichment_enabled=False, is_hierarchical=False, hierarchy_fields=None)

Bases: BaseModel

Information about a reference entity from import.yml.

Parameters:

  • name (str)

  • table_name (str)

  • kind (str)

  • description (str | None)

  • schema_fields (List[Dict[str, Any]])

  • entity_count (int | None)

  • can_enrich (bool)

  • enrichment_enabled (bool)

  • is_hierarchical (bool)

  • hierarchy_fields (HierarchyFields | None)

name: str
table_name: str
kind: str
description: str | None
schema_fields: List[Dict[str, Any]]
entity_count: int | None
can_enrich: bool
enrichment_enabled: bool
is_hierarchical: bool
hierarchy_fields: HierarchyFields | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ReferencesResponse(*, references, total)

Bases: BaseModel

Response for listing references.

Parameters:
references: List[ReferenceInfo]
total: int
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.get_references()

List all reference entities discovered from import.yml.

This endpoint dynamically discovers references (group_by targets) from the import configuration. No hardcoded entity names.

Returns:

List of references with their kind, schema, and entity count.

class niamoto.gui.api.routers.config.DatasetInfo(*, name, table_name, description=None, schema_fields=[], entity_count=None)

Bases: BaseModel

Information about a dataset entity from import.yml.

Parameters:

  • name (str)

  • table_name (str)

  • description (str | None)

  • schema_fields (List[Dict[str, Any]])

  • entity_count (int | None)

name: str
table_name: str
description: str | None
schema_fields: List[Dict[str, Any]]
entity_count: int | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.DatasetsResponse(*, datasets, total)

Bases: BaseModel

Response for listing datasets.

Parameters:
datasets: List[DatasetInfo]
total: int
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.get_reference_config(reference_name)

Get full configuration for a specific reference from import.yml.

Parameters:

reference_name (str)

async niamoto.gui.api.routers.config.update_reference_config(reference_name, config=Body(PydanticUndefined))

Update configuration for a specific reference in import.yml.

Parameters:

  • reference_name (str)

  • config (Dict[str, Any])

async niamoto.gui.api.routers.config.get_dataset_config(dataset_name)

Get full configuration for a specific dataset from import.yml.

Parameters:

dataset_name (str)

async niamoto.gui.api.routers.config.update_dataset_config(dataset_name, config=Body(PydanticUndefined))

Update configuration for a specific dataset in import.yml.

Parameters:

  • dataset_name (str)

  • config (Dict[str, Any])

async niamoto.gui.api.routers.config.get_datasets()

List all dataset entities discovered from import.yml.

Returns:

List of datasets with their schema and entity count.

async niamoto.gui.api.routers.config.get_config(config_name)

Get a configuration file by name.

Parameters:

config_name (str) – Name of the configuration (import, transform, export, config)

Returns:

The configuration content as a dictionary

async niamoto.gui.api.routers.config.update_config(config_name, update)

Update a configuration file.

Parameters:

  • config_name (str) – Name of the configuration (import, transform, export, config)

  • update (ConfigUpdate) – Configuration update with new content and backup option

Returns:

Success response with backup path if created

Return type:

ConfigResponse

async niamoto.gui.api.routers.config.validate_config(config_name, content=Body(PydanticUndefined))

Validate a configuration without saving it.

Parameters:

  • config_name (str) – Name of the configuration (import, transform, export)

  • content (Dict[str, Any]) – Configuration content to validate

Returns:

Validation result with any errors or warnings

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.list_backups(config_name)

List all backups for a configuration.

Parameters:

config_name (str) – Name of the configuration

Returns:

List of backup files with metadata

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.restore_backup(config_name, backup_filename=Body(PydanticUndefined))

Restore a configuration from a backup.

Parameters:

  • config_name (str) – Name of the configuration

  • backup_filename (str) – Name of the backup file to restore

Returns:

Success response

Return type:

ConfigResponse

class niamoto.gui.api.routers.config.ImportConfigValidateRequest(*, config)

Bases: BaseModel

Request body for validating EntityRegistry v2 import config.

Parameters:

config (str)

config: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ImportConfigValidateResponse(*, valid, errors, warnings)

Bases: BaseModel

Response for validation request.

Parameters:

  • valid (bool)

  • errors (Dict[str, list])

  • warnings (Dict[str, list])

valid: bool
errors: Dict[str, list]
warnings: Dict[str, list]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ImportConfigSaveRequest(*, config)

Bases: BaseModel

Request body for saving import config.

Parameters:

config (str)

config: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ImportConfigSaveResponse(*, success, message, path, backup_path=None)

Bases: BaseModel

Response for save request.

Parameters:

  • success (bool)

  • message (str)

  • path (str)

  • backup_path (str | None)

success: bool
message: str
path: str
backup_path: str | None
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.validate_import_v2(request)

Validate EntityRegistry v2 import configuration.

This endpoint validates the YAML structure and EntityRegistry v2 compliance without saving the configuration. It checks: - YAML syntax validity - EntityRegistry v2 structure (version, entities, datasets, references) - Entity names (unique, snake_case) - Required fields per entity type - Connector configurations - Schema field mappings - Entity links and references

Parameters:

request (ImportConfigValidateRequest) – Validation request with YAML config string

Returns:

Validation result with errors and warnings per entity

async niamoto.gui.api.routers.config.save_import_v2(request)

Save EntityRegistry v2 import configuration to import.yml.

This endpoint: 1. Validates the YAML syntax 2. Creates a backup of the existing import.yml (if it exists) 3. Writes the new configuration to config/import.yml

Parameters:

request (ImportConfigSaveRequest) – Save request with YAML config string

Returns:

Success response with file path and backup path

Raises:

HTTPException – If YAML is invalid or file operation fails

async niamoto.gui.api.routers.config.get_import_v2_schema()

Get JSON Schema for EntityRegistry v2 import configuration.

Returns a JSON Schema that describes the structure of a valid EntityRegistry v2 import.yml file. This schema can be used for: - Frontend validation - IDE autocompletion - Documentation generation

Returns:

JSON Schema object

class niamoto.gui.api.routers.config.TransformWidgetUpdate(*, plugin, params={})

Bases: BaseModel

Request body for updating a transform widget.

Parameters:

  • plugin (str)

  • params (Dict[str, Any])

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

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

class niamoto.gui.api.routers.config.ExportWidgetUpdate(*, plugin, data_source, title=None, description=None, params={})

Bases: BaseModel

Request body for updating an export widget.

Parameters:

  • plugin (str)

  • data_source (str)

  • title (str | None)

  • description (str | None)

  • params (Dict[str, Any])

plugin: str
data_source: str
title: str | None
description: 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.gui.api.routers.config.WidgetSummary(*, id, plugin, params={})

Bases: BaseModel

Summary of a widget configuration.

Parameters:

  • id (str)

  • plugin (str)

  • params (Dict[str, Any])

id: str
plugin: str
params: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.list_transform_widgets(group_by)

List all widgets in transform.yml for a specific group.

Parameters:

group_by (str) – The group name (e.g., ‘taxons’, ‘plots’, ‘shapes’)

Returns:

List of widget summaries

Return type:

List[WidgetSummary]

async niamoto.gui.api.routers.config.get_transform_widget(group_by, widget_id)

Get a specific widget from transform.yml.

Parameters:

  • group_by (str) – The group name

  • widget_id (str) – The widget identifier

Returns:

Widget summary

Return type:

WidgetSummary

async niamoto.gui.api.routers.config.update_transform_widget(group_by, widget_id, update)

Create or update a widget in transform.yml.

Parameters:

  • group_by (str) – The group name

  • widget_id (str) – The widget identifier

  • update (TransformWidgetUpdate) – Widget configuration

Returns:

Updated widget summary

Return type:

WidgetSummary

async niamoto.gui.api.routers.config.delete_transform_widget(group_by, widget_id)

Delete a widget from transform.yml.

Parameters:

  • group_by (str) – The group name

  • widget_id (str) – The widget identifier

Returns:

Success status

Return type:

Dict[str, bool]

async niamoto.gui.api.routers.config.list_export_widgets(group_by)

List all widgets in export.yml for a specific group.

Parameters:

group_by (str) – The group name (e.g., ‘taxons’, ‘plots’, ‘shapes’)

Returns:

List of widget configurations

Return type:

List[Dict[str, Any]]

async niamoto.gui.api.routers.config.update_export_widget(group_by, widget_id, update)

Create or update a widget in export.yml.

The widget_id corresponds to the data_source field.

Parameters:

  • group_by (str) – The group name

  • widget_id (str) – The widget identifier (data_source)

  • update (ExportWidgetUpdate) – Widget configuration

Returns:

Updated widget configuration

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.delete_export_widget(group_by, widget_id)

Delete a widget from export.yml.

The widget_id corresponds to the data_source field.

Parameters:

  • group_by (str) – The group name

  • widget_id (str) – The widget identifier (data_source)

Returns:

Success status

Return type:

Dict[str, bool]

class niamoto.gui.api.routers.config.IndexGeneratorPageConfigUpdate(*, title, description=None, items_per_page=24)

Bases: BaseModel

Page configuration for 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.gui.api.routers.config.IndexGeneratorFilterUpdate(*, field, values, operator='in')

Bases: BaseModel

Filter configuration for index generator.

Parameters:

  • field (str)

  • values (List[Any])

  • operator (str)

field: str
values: List[Any]
operator: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.IndexGeneratorDisplayFieldUpdate(*, name, source, fallback=None, type='text', label=None, searchable=False, format=None, mapping=None, filter_options=None, dynamic_options=False, inline_badge=False, true_label=None, false_label=None, badge_color=None, badge_style=None, badge_colors=None, badge_styles=None, tooltip_mapping=None, display=None, link_label=None, link_template=None, link_title=None, link_target=None, css_class=None, css_style=None, image_fields=None)

Bases: BaseModel

Display field configuration for 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)

  • inline_badge (bool)

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

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

  • badge_color (str | None)

  • badge_style (str | None)

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

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

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

  • display (str | None)

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

  • link_template (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
inline_badge: bool
true_label: str | Dict[str, str] | None
false_label: str | Dict[str, str] | None
badge_color: str | None
badge_style: str | None
badge_colors: Dict[str, str] | None
badge_styles: Dict[str, str] | None
tooltip_mapping: Dict[str, str] | None
display: 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.gui.api.routers.config.IndexGeneratorViewUpdate(*, type, default=False)

Bases: BaseModel

View configuration for index generator.

Parameters:

  • type (str)

  • default (bool)

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

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

class niamoto.gui.api.routers.config.IndexGeneratorConfigUpdate(*, enabled=True, template='group_index.html', page_config, filters=None, display_fields, views=None)

Bases: BaseModel

Complete index generator configuration.

Parameters:
enabled: bool
template: str
page_config: IndexGeneratorPageConfigUpdate
filters: List[IndexGeneratorFilterUpdate] | None
display_fields: List[IndexGeneratorDisplayFieldUpdate]
views: List[IndexGeneratorViewUpdate] | None
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.get_index_generator(group_by)

Get index_generator configuration for a group.

Parameters:

group_by (str) – The group name (e.g., ‘taxons’, ‘plots’, ‘shapes’)

Returns:

Index generator configuration

Raises:

HTTPException 404 – If group not found or no index_generator configured

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.update_index_generator(group_by, config)

Update index_generator configuration for a group.

Parameters:
Returns:

Updated configuration

Return type:

Dict[str, Any]

class niamoto.gui.api.routers.config.ApiExportGroupEntry(*, group_by, enabled=True)

Bases: BaseModel

Minimal info about a group inside a target.

Parameters:

  • group_by (str)

  • enabled (bool)

group_by: str
enabled: bool
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ApiExportTargetSummary(*, name, enabled=True, exporter='json_api_exporter', group_names=<factory>, groups=<factory>, params=<factory>)

Bases: BaseModel

Summary for a JSON API export target.

Parameters:

  • name (str)

  • enabled (bool)

  • exporter (str)

  • group_names (List[str])

  • groups (List[ApiExportGroupEntry])

  • params (Dict[str, Any])

name: str
enabled: bool
exporter: str
group_names: List[str]
groups: List[ApiExportGroupEntry]
params: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ApiExportTargetSettingsUpdate(*, enabled=True, params=<factory>)

Bases: BaseModel

Target-level static API settings.

Parameters:

  • enabled (bool)

  • params (Dict[str, Any])

enabled: bool
params: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ApiExportTargetCreate(*, name, template, params=<factory>)

Bases: BaseModel

Create a new JSON API export target.

Parameters:

  • name (str)

  • template (Literal['simple', 'dwc'])

  • params (Dict[str, Any])

name: str
template: Literal['simple', 'dwc']
params: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.ApiExportGroupConfigUpdate(*, enabled=True, data_source=None, detail=None, index=None, json_options=None, transformer_plugin=None, transformer_params=None)

Bases: BaseModel

Per-group configuration for a static API export target.

Parameters:

  • enabled (bool)

  • data_source (str | None)

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

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

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

  • transformer_plugin (str | None)

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

enabled: bool
data_source: str | None
detail: Dict[str, Any] | None
index: Dict[str, Any] | None
json_options: Dict[str, Any] | None
transformer_plugin: str | None
transformer_params: Dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.list_api_export_targets()

List all configured static API export targets.

Return type:

List[ApiExportTargetSummary]

async niamoto.gui.api.routers.config.create_api_export_target(body)

Create a new JSON API export target from a template.

Parameters:

body (ApiExportTargetCreate)

Return type:

ApiExportTargetSummary

async niamoto.gui.api.routers.config.get_api_export_target_settings(export_name)

Get global settings for a static API export target.

Parameters:

export_name (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.update_api_export_target_settings(export_name, config)

Update target-level settings for a static API export target.

Parameters:
Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.get_api_export_group_config(export_name, group_by)

Get per-group settings for a specific static API export target.

Parameters:

  • export_name (str)

  • group_by (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.config.update_api_export_group_config(export_name, group_by, config)

Update per-group settings for a static API export target.

Parameters:
Return type:

Dict[str, Any]

class niamoto.gui.api.routers.config.SuggestedDisplayField(*, name, source, type, label, searchable=False, cardinality=None, sample_values=None, suggested_as_filter=False, format=None, dynamic_options=False, priority='high', display=None, inline_badge=False, link_label=None, link_title=None, link_target=None, image_fields=None)

Bases: BaseModel

Suggested display field configuration.

Parameters:

  • name (str)

  • source (str)

  • type (str)

  • label (str)

  • searchable (bool)

  • cardinality (int | None)

  • sample_values (List[str] | None)

  • suggested_as_filter (bool)

  • format (str | None)

  • dynamic_options (bool)

  • priority (str)

  • display (str | None)

  • inline_badge (bool)

  • link_label (str | None)

  • link_title (str | None)

  • link_target (str | None)

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

name: str
source: str
type: str
label: str
searchable: bool
cardinality: int | None
sample_values: List[str] | None
suggested_as_filter: bool
format: str | None
dynamic_options: bool
priority: str
display: str | None
inline_badge: bool
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.gui.api.routers.config.SuggestedFilter(*, field, source, label, type, values, operator='in')

Bases: BaseModel

Suggested filter configuration.

Parameters:

  • field (str)

  • source (str)

  • label (str)

  • type (str)

  • values (List[Any])

  • operator (str)

field: str
source: str
label: str
type: str
values: List[Any]
operator: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.config.IndexFieldSuggestions(*, display_fields, filters, total_entities)

Bases: BaseModel

Response with suggested fields for index generator.

Parameters:
display_fields: List[SuggestedDisplayField]
filters: List[SuggestedFilter]
total_entities: int
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.config.suggest_api_export_index_fields(export_name, group_by)

Reuse index field suggestions for static API export indexes.

Parameters:

  • export_name (str)

  • group_by (str)

Return type:

IndexFieldSuggestions

async niamoto.gui.api.routers.config.suggest_index_fields(group_by)

Analyze transformed data and suggest display fields and filters.

This endpoint: 1. First tries to load the transformed stats table (if transform was already run) 2. If not available, infers the schema from transform.yml configuration 3. Uses source data to detect field types and sample values 4. Suggests fields for display and filtering based on cardinality

Parameters:

group_by (str) – The group name (e.g., ‘taxons’, ‘plots’, ‘shapes’)

Returns:

Suggested display_fields and filters

Return type:

IndexFieldSuggestions

async niamoto.gui.api.routers.config.scaffold_configs_endpoint()

Scaffold les configs transform.yml et export.yml à partir de import.yml.

Crée des groupes minimaux pour chaque référence absente. Idempotent : les groupes existants ne sont pas modifiés. Utile après ajout d’une nouvelle référence dans import.yml.

Return type:

Dict[str, Any]

niamoto.gui.api.routers.data_explorer module

niamoto.gui.api.routers.database module

Database introspection API endpoints.

class niamoto.gui.api.routers.database.ColumnInfo(*, name, type, nullable, primary_key=False, foreign_key=None)

Bases: BaseModel

Database column information.

Parameters:

  • name (str)

  • type (str)

  • nullable (bool)

  • primary_key (bool)

  • foreign_key (str | None)

name: str
type: str
nullable: bool
primary_key: bool
foreign_key: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.database.TableInfo(*, name, row_count, columns, indexes=[], is_view=False)

Bases: BaseModel

Database table information.

Parameters:

  • name (str)

  • row_count (int)

  • columns (List[ColumnInfo])

  • indexes (List[str])

  • is_view (bool)

name: str
row_count: int
columns: List[ColumnInfo]
indexes: List[str]
is_view: bool
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.database.DatabaseSchema(*, tables, views=[], total_size=None)

Bases: BaseModel

Complete database schema information.

Parameters:
tables: List[TableInfo]
views: List[TableInfo]
total_size: int | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.database.TablePreview(*, table_name, columns, rows, total_rows, preview_limit)

Bases: BaseModel

Preview of table data.

Parameters:

  • table_name (str)

  • columns (List[str])

  • rows (List[Dict[str, Any]])

  • total_rows (int)

  • preview_limit (int)

table_name: str
columns: List[str]
rows: List[Dict[str, Any]]
total_rows: int
preview_limit: int
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.database.TableStats(*, table_name, row_count, column_count, null_counts, unique_counts, data_types)

Bases: BaseModel

Statistics for a table.

Parameters:

  • table_name (str)

  • row_count (int)

  • column_count (int)

  • null_counts (Dict[str, int])

  • unique_counts (Dict[str, int])

  • data_types (Dict[str, str])

table_name: str
row_count: int
column_count: int
null_counts: Dict[str, int]
unique_counts: Dict[str, int]
data_types: Dict[str, str]
model_config: ClassVar[ConfigDict] = {}

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

niamoto.gui.api.routers.database.get_database_path()

Get the path to the configured analytics database (SQLite or DuckDB).

Return type:

Path | None

async niamoto.gui.api.routers.database.get_database_schema()

Get the complete database schema including tables, views, and columns.

Returns detailed information about all tables and views in the database.

async niamoto.gui.api.routers.database.get_table_preview(table_name, limit=Query(100), offset=Query(0))

Get a preview of data from a specific table.

Parameters:

  • table_name (str) – Name of the table to preview

  • limit (int) – Maximum number of rows to return (1-1000)

  • offset (int) – Number of rows to skip for pagination

Returns:

Preview of the table data with column names and sample rows

async niamoto.gui.api.routers.database.get_table_stats(table_name)

Get statistics for a specific table.

Parameters:

table_name (str) – Name of the table to analyze

Returns:

Statistical information about the table including null counts, unique value counts, and data types for each column

async niamoto.gui.api.routers.database.execute_query(query=Query(PydanticUndefined), limit=Query(100))

Execute a custom SQL query (read-only).

Parameters:

  • query (str) – SQL query to execute (SELECT only)

  • limit (int) – Maximum number of rows to return

Returns:

Query results with column names and data

Note: Only SELECT queries are allowed for safety

niamoto.gui.api.routers.deploy module

Deploy router for the Niamoto GUI API.

Supports deployment to any platform registered as a deployer plugin. Built-in: Cloudflare Workers, GitHub Pages, Netlify, Vercel, Render, SSH/SFTP.

class niamoto.gui.api.routers.deploy.CredentialSaveRequest(*, key, value)

Bases: BaseModel

Parameters:

  • key (str)

  • value (str)

key: str
value: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.deploy.DeployRequest(*, platform, project_name, branch=None, extra={})

Bases: BaseModel

Parameters:

  • platform (str)

  • project_name (str)

  • branch (str | None)

  • extra (dict[str, Any])

platform: str
project_name: str
branch: str | None
extra: dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.deploy.list_platforms()

List all registered deployer platforms.

async niamoto.gui.api.routers.deploy.save_credential(platform, request)

Save a credential to the OS keyring.

Parameters:
async niamoto.gui.api.routers.deploy.check_credentials(platform)

Check if a platform has credentials configured.

Parameters:

platform (str)

async niamoto.gui.api.routers.deploy.delete_credential(platform, key)

Delete a credential from the OS keyring.

Parameters:

  • platform (str)

  • key (str)

async niamoto.gui.api.routers.deploy.validate_credentials(platform)

Validate credentials by making a test API call.

Parameters:

platform (str)

async niamoto.gui.api.routers.deploy.deploy(request)

Deploy to any platform with SSE streaming logs.

Parameters:

request (DeployRequest)

async niamoto.gui.api.routers.deploy.validate_exports(request)

Validate exports directory before deployment (dry run).

Parameters:

request (DeployRequest)

async niamoto.gui.api.routers.deploy.check_site_health(url=Query(PydanticUndefined))

Check if a deployed site is online.

Performs a GET request and follows both HTTP and HTML meta-refresh redirects to catch cases where the landing page returns 200 but redirects to a 404 page (common with GitHub Pages after branch deletion).

Parameters:

url (str)

class niamoto.gui.api.routers.deploy.UnpublishRequest(*, platform, project_name, branch=None, extra={})

Bases: BaseModel

Parameters:

  • platform (str)

  • project_name (str)

  • branch (str | None)

  • extra (dict[str, Any])

platform: str
project_name: str
branch: str | None
extra: dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.deploy.unpublish(request)

Unpublish a site from a platform with SSE streaming logs.

Parameters:

request (UnpublishRequest)

niamoto.gui.api.routers.entities module

Entities API endpoints for accessing entity data with transformations and EntityRegistry.

class niamoto.gui.api.routers.entities.EntitySummary(*, id, name, display_name=None)

Bases: BaseModel

Summary information about an entity.

Parameters:

  • id (str)

  • name (str)

  • display_name (str | None)

id: str
name: str
display_name: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.entities.EntityDetail(*, id, name, group_by, widgets_data)

Bases: BaseModel

Detailed entity information including widgets_data.

Parameters:

  • id (str)

  • name (str)

  • group_by (str)

  • widgets_data (Dict[str, Any])

id: str
name: str
group_by: str
widgets_data: Dict[str, Any]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.entities.TransformationPreview(*, entity_id, entity_name, group_by, transformation_key, transformation_data, widget_plugin=None)

Bases: BaseModel

Preview of a specific transformation and its widget.

Parameters:

  • entity_id (str)

  • entity_name (str)

  • group_by (str)

  • transformation_key (str)

  • transformation_data (Dict[str, Any])

  • widget_plugin (str | None)

entity_id: str
entity_name: str
group_by: str
transformation_key: str
transformation_data: Dict[str, Any]
widget_plugin: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.entities.EntityInfo(*, name, kind, entity_type)

Bases: BaseModel

Information about a registered entity from EntityRegistry.

Parameters:

  • name (str)

  • kind (str)

  • entity_type (str)

name: str
kind: str
entity_type: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.entities.EntityListResponse(*, datasets=[], references=[], all=[])

Bases: BaseModel

Response with available entities grouped by kind.

Parameters:

  • datasets (List[str])

  • references (List[str])

  • all (List[EntityInfo])

datasets: List[str]
references: List[str]
all: List[EntityInfo]
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.entities.get_available_entities(kind=Query(None))

Get list of available entities from EntityRegistry.

This endpoint is used by entity-select widgets in plugin forms to dynamically populate entity dropdowns.

Parameters:

kind (str | None) – Optional filter - ‘dataset’ or ‘reference’

Returns:

EntityListResponse with entities grouped by kind

async niamoto.gui.api.routers.entities.list_entities(group_by, limit=None)

List entities for a specific group_by (taxon, plot, shape).

Parameters:

  • group_by (str) – Entity type (taxon, plot, or shape)

  • limit (int | None) – Maximum number of entities to return

Returns:

List of entity summaries

async niamoto.gui.api.routers.entities.get_entity_detail(group_by, entity_id)

Get detailed information about a specific entity including all widgets_data.

Parameters:

  • group_by (str) – Entity type (taxon, plot, or shape)

  • entity_id (str) – ID of the entity (as string to preserve large integer precision)

Returns:

Entity details with widgets_data

async niamoto.gui.api.routers.entities.get_transformation_preview(group_by, entity_id, transform_key)

Get preview of a specific transformation for an entity.

Parameters:

  • group_by (str) – Entity type (taxon, plot, or shape)

  • entity_id (str) – ID of the entity (as string to preserve large integer precision)

  • transform_key (str) – Key of the transformation (e.g., ‘dbh_distribution’)

Returns:

Transformation preview with data

async niamoto.gui.api.routers.entities.render_widget(group_by, entity_id, transform_key)

Render a widget HTML for a specific transformation.

Parameters:

  • group_by (str) – Entity type (taxon, plot, or shape)

  • entity_id (str) – ID of the entity (as string to preserve large integer precision)

  • transform_key (str) – Key of the transformation (e.g., ‘dbh_distribution’)

Returns:

HTML content of the rendered widget

niamoto.gui.api.routers.export module

Export API endpoints for generating static sites and exports.

class niamoto.gui.api.routers.export.ExportRequest(*, config_path='config/export.yml', export_types=None, include_transform=False)

Bases: BaseModel

Request model for executing exports.

Parameters:

  • config_path (str | None)

  • export_types (List[str] | None)

  • include_transform (bool)

config_path: str | None
export_types: List[str] | None
include_transform: bool
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.export.ExportResponse(*, job_id, status, message, started_at)

Bases: BaseModel

Response model for export execution.

Parameters:

  • job_id (str)

  • status (str)

  • message (str)

  • started_at (datetime)

job_id: str
status: str
message: str
started_at: datetime
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.export.ExportStatus(*, job_id, status, progress, message, phase=None, started_at, completed_at=None, result=None, error=None)

Bases: BaseModel

Status model for export jobs.

Parameters:

  • job_id (str)

  • status (str)

  • progress (int)

  • message (str)

  • phase (str | None)

  • started_at (datetime)

  • completed_at (datetime | None)

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

  • error (str | None)

job_id: str
status: str
progress: int
message: str
phase: str | None
started_at: datetime
completed_at: datetime | None
result: Dict[str, Any] | None
error: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.export.ExportMetrics(*, total_exports, completed_exports, failed_exports, generated_pages, static_site_path, execution_time)

Bases: BaseModel

Metrics for completed exports.

Parameters:

  • total_exports (int)

  • completed_exports (int)

  • failed_exports (int)

  • generated_pages (int)

  • static_site_path (str | None)

  • execution_time (float)

total_exports: int
completed_exports: int
failed_exports: int
generated_pages: int
static_site_path: str | None
execution_time: float
model_config: ClassVar[ConfigDict] = {}

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

niamoto.gui.api.routers.export.get_export_config(config_path)

Load and parse export configuration.

Parameters:

config_path (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.export.execute_export_background(job_id, job_store, config_path, export_types=None, include_transform=False)

Execute exports in the background, optionally preceded by transform.

Parameters:

  • job_id (str)

  • job_store (JobFileStore)

  • config_path (str)

  • export_types (List[str] | None)

  • include_transform (bool)

async niamoto.gui.api.routers.export.execute_export(request, background_tasks, http_request)

Execute exports based on configuration.

This starts a background job that processes the exports defined in the export.yml configuration file.

Parameters:

  • request (ExportRequest)

  • background_tasks (BackgroundTasks)

  • http_request (Request)

async niamoto.gui.api.routers.export.get_export_status(job_id, http_request)

Get the status of an export job.

Returns the current status, progress, and result (if completed) of the specified export job.

Parameters:

  • job_id (str)

  • http_request (Request)

async niamoto.gui.api.routers.export.list_export_jobs(http_request)

List all export jobs.

Returns a list of all export jobs with their current status.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.export.cancel_export_job(job_id, http_request)

Cancel a running export job.

Parameters:

  • job_id (str)

  • http_request (Request)

async niamoto.gui.api.routers.export.clear_export_history(http_request)

Clear persisted export history for the current project.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.export.get_export_config_endpoint()

Get the current export configuration.

Returns the parsed content of the export.yml file.

async niamoto.gui.api.routers.export.get_active_export_job(http_request)

Get the currently active job (running or recently completed).

Returns the active job or null if none.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.export.get_export_metrics(http_request)

Get metrics from the last completed export.

Returns statistics about the exports performed.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.export.execute_export_cli(background_tasks, http_request)

Execute export using the Niamoto CLI command.

This runs ‘niamoto export’ in the background and returns immediately.

Parameters:

  • background_tasks (BackgroundTasks)

  • http_request (Request)

niamoto.gui.api.routers.files module

File management API endpoints.

class niamoto.gui.api.routers.files.ApiTestRequest(*, url, headers={}, params={})

Bases: BaseModel

Request model for API testing.

Parameters:

  • url (str)

  • headers (Dict[str, str])

  • params (Dict[str, str])

url: str
headers: Dict[str, str]
params: Dict[str, str]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.files.ApiTestResponse(*, success, data=None, error=None)

Bases: BaseModel

Response model for API testing.

Parameters:

  • success (bool)

  • data (Any | None)

  • error (str | None)

success: bool
data: Any | None
error: str | None
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.files.analyze_file(file=File(PydanticUndefined), entity_type=Form(PydanticUndefined))

Analyze a file for import configuration.

Parameters:

  • file (UploadFile)

  • entity_type (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.analyze_geopackage(content, filename)

Analyze GeoPackage file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

niamoto.gui.api.routers.files.get_field_suggestions(analysis, import_type)

Get field mapping suggestions based on import type and file analysis.

Parameters:

  • analysis (Dict[str, Any])

  • import_type (str)

Return type:

Dict[str, List[str]]

niamoto.gui.api.routers.files.find_matching_columns(lower_columns, patterns)

Find columns that match any of the given patterns.

Parameters:

  • lower_columns (Dict[str, str])

  • patterns (List[str])

Return type:

List[str]

async niamoto.gui.api.routers.files.test_api_connection(request)

Test an API connection with provided configuration.

Parameters:

request (ApiTestRequest)

Return type:

ApiTestResponse

niamoto.gui.api.routers.files.suggest_shape_mappings(columns)

Suggest mappings for shape attributes.

Parameters:

columns (List[str])

Return type:

Dict[str, List[str]]

async niamoto.gui.api.routers.files.analyze_shape(content, filename)

Analyze shapefile or other spatial file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.analyze_csv(content, filename)

Analyze CSV file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.analyze_excel(content, filename)

Analyze Excel file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.analyze_json(content, filename)

Analyze JSON file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.analyze_geojson(content, filename)

Analyze GeoJSON file content.

Parameters:

  • content (bytes)

  • filename (str)

Return type:

Dict[str, Any]

niamoto.gui.api.routers.files.infer_column_type(values)

Infer the data type of a column from sample values.

Parameters:

values (List[str])

Return type:

str

async niamoto.gui.api.routers.files.browse_files(path='.')

Browse files in the filesystem.

Parameters:

path (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.list_exports()

List all exported files organized by type.

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.read_export_file(file_path)

Read content of an exported file.

Parameters:

file_path (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.get_exports_structure()

Get the directory structure of exports folder.

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.files.serve_file(file_path)

Serve a file from the project directory.

Used for displaying images like logos in the UI.

Parameters:

file_path (str)

niamoto.gui.api.routers.imports module

Generic import API endpoints using entity registry and typed configurations.

class niamoto.gui.api.routers.imports.ImportStatus(*, entity_name, entity_type, is_imported, row_count=0)

Bases: BaseModel

Status of a particular entity import.

Parameters:

  • entity_name (str)

  • entity_type (str)

  • is_imported (bool)

  • row_count (int)

entity_name: str
entity_type: str
is_imported: bool
row_count: int
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.imports.ImportStatusResponse(*, references=[], datasets=[])

Bases: BaseModel

Response containing status of all imports.

Parameters:
references: List[ImportStatus]
datasets: List[ImportStatus]
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.imports.ImportJobResponse(*, job_id, status, created_at, message)

Bases: BaseModel

Response model for import job creation.

Parameters:

  • job_id (str)

  • status (str)

  • created_at (str)

  • message (str)

job_id: str
status: str
created_at: str
message: str
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.imports.execute_import_all(background_tasks, reset_table=Form(False))

Execute import of all entities from generic configuration.

Parameters:

  • background_tasks (BackgroundTasks)

  • reset_table (bool)

Return type:

ImportJobResponse

async niamoto.gui.api.routers.imports.execute_import_reference(entity_name, background_tasks, reset_table=Form(False))

Execute import of a specific reference entity.

Parameters:

  • entity_name (str)

  • background_tasks (BackgroundTasks)

  • reset_table (bool)

Return type:

ImportJobResponse

async niamoto.gui.api.routers.imports.execute_import_dataset(entity_name, background_tasks, reset_table=Form(False))

Execute import of a specific dataset entity.

Parameters:

  • entity_name (str)

  • background_tasks (BackgroundTasks)

  • reset_table (bool)

Return type:

ImportJobResponse

async niamoto.gui.api.routers.imports.get_job_status(job_id)

Get the status of an import job.

Parameters:

job_id (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.imports.list_import_jobs(limit=10, offset=0, status=None)

List all import jobs with optional filtering.

Parameters:

  • limit (int)

  • offset (int)

  • status (str | None)

Return type:

Dict[str, Any]

class niamoto.gui.api.routers.imports.DeleteEntityRequest(*, delete_table=False)

Bases: BaseModel

Request to delete an entity.

Parameters:

delete_table (bool)

delete_table: bool
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.imports.delete_entity(entity_type, entity_name, delete_table=False)

Delete an entity from import.yml configuration.

Parameters:

  • entity_type (str) – Type of entity (‘dataset’ or ‘reference’)

  • entity_name (str) – Name of the entity to delete

  • delete_table (bool) – If True, also drop the associated database table

Returns:

Success message

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.imports.list_entities()

List all entities defined in import.yml configuration with their actual table names.

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.imports.get_import_status()

Check which entities have been imported and their row counts.

Return type:

ImportStatusResponse

async niamoto.gui.api.routers.imports.process_generic_import_all(job_id, reset_table)

Process generic import of all entities in background.

Parameters:

  • job_id (str)

  • reset_table (bool)

async niamoto.gui.api.routers.imports.process_generic_import_entity(job_id, entity_name, entity_type, reset_table)

Process generic import of a single entity in background.

Parameters:

  • job_id (str)

  • entity_name (str)

  • entity_type (str)

  • reset_table (bool)

class niamoto.gui.api.routers.imports.ImpactCheckRequest(*, file_path)

Bases: BaseModel

Parameters:

file_path (str)

file_path: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.imports.ImpactItemResponse(*, column, level, detail, referenced_in=[], old_type=None, new_type=None)

Bases: BaseModel

Parameters:

  • column (str)

  • level (str)

  • detail (str)

  • referenced_in (List[str])

  • old_type (str | None)

  • new_type (str | None)

column: str
level: str
detail: str
referenced_in: List[str]
old_type: str | None
new_type: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.imports.ColumnMatchResponse(*, name, old_type, new_type)

Bases: BaseModel

Parameters:

  • name (str)

  • old_type (str)

  • new_type (str)

name: str
old_type: str
new_type: str
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.imports.ImpactCheckResponse(*, entity_name=None, matched_columns=[], impacts=[], error=None, skipped_reason=None, info_message=None, has_blockers=False, has_warnings=False, has_opportunities=False)

Bases: BaseModel

Parameters:

  • entity_name (str | None)

  • matched_columns (List[ColumnMatchResponse])

  • impacts (List[ImpactItemResponse])

  • error (str | None)

  • skipped_reason (str | None)

  • info_message (str | None)

  • has_blockers (bool)

  • has_warnings (bool)

  • has_opportunities (bool)

entity_name: str | None
matched_columns: List[ColumnMatchResponse]
impacts: List[ImpactItemResponse]
error: str | None
skipped_reason: str | None
info_message: str | None
has_blockers: bool
has_warnings: bool
has_opportunities: bool
model_config: ClassVar[ConfigDict] = {}

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

async niamoto.gui.api.routers.imports.impact_check(request)

Check compatibility between a source file and existing configuration.

Resolves the entity from the file path basename, then runs the impact check against import.yml + transform.yml.

Parameters:

request (ImpactCheckRequest)

niamoto.gui.api.routers.plugins module

Plugin registry API endpoints using the real Niamoto plugin system.

class niamoto.gui.api.routers.plugins.ParameterSchema(*, name, type, required=False, default=None, description=None, enum=None, min=None, max=None)

Bases: BaseModel

Schema for a plugin parameter.

Parameters:

  • name (str)

  • type (str)

  • required (bool)

  • default (Any)

  • description (str | None)

  • enum (List[Any] | None)

  • min (float | None)

  • max (float | None)

name: str
type: str
required: bool
default: Any
description: str | None
enum: List[Any] | None
min: float | None
max: float | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.plugins.PluginInfo(*, id, name, type, description, version='1.0.0', author=None, category=None, parameters_schema=[], compatible_inputs=[], output_format=None, example_config=None)

Bases: BaseModel

Information about a plugin.

Parameters:

  • id (str)

  • name (str)

  • type (str)

  • description (str)

  • version (str | None)

  • author (str | None)

  • category (str | None)

  • parameters_schema (List[ParameterSchema])

  • compatible_inputs (List[str])

  • output_format (str | None)

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

id: str
name: str
type: str
description: str
version: str | None
author: str | None
category: str | None
parameters_schema: List[ParameterSchema]
compatible_inputs: List[str]
output_format: str | None
example_config: Dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.plugins.CompatibilityCheck(*, source_data, plugin_id, config=None)

Bases: BaseModel

Request body for compatibility check.

Parameters:

  • source_data (Dict[str, Any])

  • plugin_id (str)

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

source_data: Dict[str, Any]
plugin_id: str
config: Dict[str, Any] | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.plugins.CompatibilityResult(*, compatible, reason=None, suggestions=[])

Bases: BaseModel

Result of compatibility check.

Parameters:

  • compatible (bool)

  • reason (str | None)

  • suggestions (List[str])

compatible: bool
reason: str | None
suggestions: List[str]
model_config: ClassVar[ConfigDict] = {}

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

niamoto.gui.api.routers.plugins.load_all_plugins()

Load all plugins from the plugin directories to populate the registry. This ensures all available plugins are registered.

niamoto.gui.api.routers.plugins.get_plugin_info_from_class(name, plugin_class)

Extract plugin information from a plugin class.

Parameters:

  • name (str) – The registered name of the plugin

  • plugin_class (type) – The plugin class

Returns:

PluginInfo object with extracted information

Return type:

PluginInfo

async niamoto.gui.api.routers.plugins.list_plugins(type=Query(None), category=Query(None), compatible_with=Query(None))

Get list of available plugins from the real plugin registry.

Parameters:

  • type (str | None) – Filter by plugin type (loader, transformer, exporter, widget)

  • category (str | None) – Filter by category

  • compatible_with (str | None) – Filter by compatible input format

Returns:

List of plugins matching the filters

async niamoto.gui.api.routers.plugins.get_plugin(plugin_id)

Get detailed information about a specific plugin.

Parameters:

plugin_id (str) – ID of the plugin

Returns:

Detailed plugin information

async niamoto.gui.api.routers.plugins.list_categories()

Get list of all plugin categories.

Returns:

List of unique categories across all plugins

async niamoto.gui.api.routers.plugins.list_plugin_types()

Get list of all plugin types.

Returns:

List of available plugin types

async niamoto.gui.api.routers.plugins.check_compatibility(check)

Check if a plugin is compatible with given source data.

Parameters:

check (CompatibilityCheck) – Compatibility check request with source data and plugin ID

Returns:

Compatibility result with explanation

async niamoto.gui.api.routers.plugins.get_plugin_json_schema(plugin_id)

Get the full JSON schema for a plugin’s parameters.

This endpoint returns the complete Pydantic-generated JSON schema including all UI hints from json_schema_extra.

Parameters:

plugin_id (str) – ID of the plugin

Returns:

Complete JSON schema for the plugin parameters

niamoto.gui.api.routers.transform module

Transform API endpoints for executing data transformations.

class niamoto.gui.api.routers.transform.TransformRequest(*, config_path='config/transform.yml', transformations=None, group_by=None, group_bys=None)

Bases: BaseModel

Request model for executing transformations.

Parameters:

  • config_path (str | None)

  • transformations (List[str] | None)

  • group_by (str | None)

  • group_bys (List[str] | None)

config_path: str | None
transformations: List[str] | None
group_by: str | None
group_bys: List[str] | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.transform.TransformResponse(*, job_id, status, message, started_at)

Bases: BaseModel

Response model for transform execution.

Parameters:

  • job_id (str)

  • status (str)

  • message (str)

  • started_at (datetime)

job_id: str
status: str
message: str
started_at: datetime
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.transform.TransformStatus(*, job_id, status, progress, message, phase=None, group_by=None, group_bys=None, started_at, completed_at=None, result=None, error=None)

Bases: BaseModel

Status model for transform jobs.

Parameters:

  • job_id (str)

  • status (str)

  • progress (int)

  • message (str)

  • phase (str | None)

  • group_by (str | None)

  • group_bys (List[str] | None)

  • started_at (datetime)

  • completed_at (datetime | None)

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

  • error (str | None)

job_id: str
status: str
progress: int
message: str
phase: str | None
group_by: str | None
group_bys: List[str] | None
started_at: datetime
completed_at: datetime | None
result: Dict[str, Any] | None
error: str | None
model_config: ClassVar[ConfigDict] = {}

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

class niamoto.gui.api.routers.transform.TransformMetrics(*, total_transformations, completed_transformations, failed_transformations, total_widgets, generated_files, execution_time)

Bases: BaseModel

Metrics for completed transformations.

Parameters:

  • total_transformations (int)

  • completed_transformations (int)

  • failed_transformations (int)

  • total_widgets (int)

  • generated_files (List[str])

  • execution_time (float)

total_transformations: int
completed_transformations: int
failed_transformations: int
total_widgets: int
generated_files: List[str]
execution_time: float
model_config: ClassVar[ConfigDict] = {}

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

niamoto.gui.api.routers.transform.get_transform_config(config_path)

Load and parse transform configuration.

Parameters:

config_path (str)

Return type:

Dict[str, Any]

async niamoto.gui.api.routers.transform.execute_transform_background(job_id, job_store, config_path, transformations=None, group_by=None, group_bys=None)

Execute transformations in the background.

Parameters:

  • job_id (str)

  • job_store (JobFileStore)

  • config_path (str)

  • transformations (List[str] | None)

  • group_by (str | None)

  • group_bys (List[str] | None)

async niamoto.gui.api.routers.transform.execute_transform(request, background_tasks, http_request)

Execute data transformations based on configuration.

This starts a background job that processes the transformations defined in the transform.yml configuration file.

Parameters:

  • request (TransformRequest)

  • background_tasks (BackgroundTasks)

  • http_request (Request)

async niamoto.gui.api.routers.transform.get_transform_status(job_id, http_request)

Get the status of a transform job.

Returns the current status, progress, and result (if completed) of the specified transform job.

Parameters:

  • job_id (str)

  • http_request (Request)

async niamoto.gui.api.routers.transform.list_transform_jobs(http_request)

List all transform jobs.

Returns a list of all transform jobs with their current status.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.transform.cancel_transform_job(job_id, http_request)

Cancel a running transform job.

Parameters:

  • job_id (str)

  • http_request (Request)

async niamoto.gui.api.routers.transform.get_transform_config_endpoint()

Get the current transform configuration.

Returns the parsed content of the transform.yml file.

async niamoto.gui.api.routers.transform.get_transform_metrics(http_request)

Get metrics from the last completed transform.

Returns statistics about the transformations performed.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.transform.get_active_transform_job(http_request)

Get the currently active job (running or recently completed).

Returns the active job or null if none.

Parameters:

http_request (Request)

async niamoto.gui.api.routers.transform.get_last_transform_run(group_by, http_request)

Get the last completed transform for a specific group.

Returns the last terminal job for this group_by or null.

Parameters:

  • group_by (str)

  • http_request (Request)

async niamoto.gui.api.routers.transform.get_transform_sources(group_by=Query(None))

Get available transform sources from transform.yml.

Transform sources are intermediate data files loaded via stats_loader that are used by class_objects plugins.

Parameters:

group_by (str | None) – Optional filter - return only sources for this specific group_by

Returns:

Dictionary with ‘sources’ key containing list of source names

Return type:

Dict[str, List[str]]

Example

GET /api/transform/sources?group_by=shapes Returns: {“sources”: [“raw_shape_stats”, “shape_stats”]}

Module contents

API routers for Niamoto GUI.