niamoto.gui.api.routers package

Submodules

niamoto.gui.api.routers.bootstrap module

niamoto.gui.api.routers.config module

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

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

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 (Any)

  • widget_plugin (str | None)

entity_id: str
entity_name: str
group_by: str
transformation_key: str
transformation_data: 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 (Literal['dataset', 'reference'] | 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].

class niamoto.gui.api.routers.export.ExportExecutionContext(work_dir, config_path, db_path)

Bases: object

Immutable project context captured when the export request starts.

Parameters:

  • work_dir (Path)

  • config_path (Path)

  • db_path (Path | None)

work_dir: Path
config_path: Path
db_path: Path | None
niamoto.gui.api.routers.export.get_config_path(config_file)

Resolve an export config path for callers that patch this hook in tests.

Parameters:

config_file (str | Path)

Return type:

Path

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

Load and parse export configuration.

Parameters:

config_path (str | Path)

Return type:

Dict[str, Any]

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

Execute exports in the background, optionally preceded by transform.

Parameters:

  • job_id (str)

  • job_store (JobFileStore)

  • execution_context (ExportExecutionContext)

  • 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(request, background_tasks, reset_table=Form(False))

Execute import of all entities from generic configuration.

Parameters:

  • request (Request)

  • background_tasks (BackgroundTasks)

  • reset_table (bool)

Return type:

ImportJobResponse

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

Execute import of a specific reference entity.

Parameters:

  • request (Request)

  • entity_name (str)

  • background_tasks (BackgroundTasks)

  • reset_table (bool)

Return type:

ImportJobResponse

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

Execute import of a specific dataset entity.

Parameters:

  • request (Request)

  • 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=Query(10), offset=Query(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(request, 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

  • request (Request)

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, working_directory=None)

Process generic import of all entities in background.

Parameters:

  • job_id (str)

  • reset_table (bool)

  • working_directory (str | None)

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

Process generic import of a single entity in background.

Parameters:

  • job_id (str)

  • entity_name (str)

  • entity_type (str)

  • reset_table (bool)

  • working_directory (str | None)

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=<factory>, 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.WidgetImpactResponse(*, widget_id, collection, status, detail, affected_columns=<factory>, transformer_plugin=None, widget_plugin=None)

Bases: BaseModel

Parameters:

  • widget_id (str)

  • collection (str)

  • status (Literal['still_valid', 'degraded', 'broken', 'newly_available', 'unknown'])

  • detail (str)

  • affected_columns (List[str])

  • transformer_plugin (str | None)

  • widget_plugin (str | None)

widget_id: str
collection: str
status: Literal['still_valid', 'degraded', 'broken', 'newly_available', 'unknown']
detail: str
affected_columns: List[str]
transformer_plugin: str | None
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.imports.ImpactCheckResponse(*, entity_name=None, matched_columns=<factory>, impacts=<factory>, error=None, skipped_reason=None, info_message=None, has_blockers=False, has_warnings=False, has_opportunities=False, widget_impacts=<factory>, widget_impact_summary=<factory>, widget_repair_context=<factory>)

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)

  • widget_impacts (List[WidgetImpactResponse])

  • widget_impact_summary (Dict[str, int])

  • widget_repair_context (Dict[str, Any])

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
widget_impacts: List[WidgetImpactResponse]
widget_impact_summary: Dict[str, int]
widget_repair_context: 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.imports.impact_check(request)

Check compatibility between a source file and existing configuration.

Resolves the entity from the project-relative file path, 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 available plugins through the configured project/user/system cascade.

The registry is rebuilt from the cascade so project-local plugins can override lower-priority plugins. If discovery fails, keep the previous registry intact for in-flight API requests.

Return type:

None

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

Extract plugin information from a plugin class.

Parameters:

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

  • plugin_class (type) – The plugin class

  • plugin_type (PluginType | None)

Returns:

PluginInfo object with extracted information

Return type:

PluginInfo

async niamoto.gui.api.routers.plugins.list_plugins(request, 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

  • request (Request)

Returns:

List of plugins matching the filters

async niamoto.gui.api.routers.plugins.get_plugin_json_schema_priority_route(plugin_id, request)

Keep plugin schema lookup ahead of the generic plugin detail route.

Parameters:

  • plugin_id (str)

  • request (Request)

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

Get list of all plugin categories.

Returns:

List of unique categories across all plugins

Parameters:

request (Request)

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, request)

Check if a plugin is compatible with given source data.

Parameters:

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

  • request (Request)

Returns:

Compatibility result with explanation

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

Get detailed information about a specific plugin.

Parameters:

  • plugin_id (str) – ID of the plugin

  • request (Request)

Returns:

Detailed plugin information

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

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

  • request (Request)

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)

  • transformations (List[str] | None)

  • group_by (str | None)

  • group_bys (List[str] | None)

config_path: str
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].

exception niamoto.gui.api.routers.transform.TransformCancelled

Bases: RuntimeError

Raised inside the transform worker when a user cancels the job.

niamoto.gui.api.routers.transform.get_transform_config(config_path, working_directory=None)

Load and parse transform configuration.

Parameters:

  • config_path (str)

  • working_directory (Path | None)

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, working_directory=None, database_path=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)

  • working_directory (str | None)

  • database_path (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.