pixano.app.routers.utils

assert_table_in_group(dataset, table, group)

Assert that a table belongs to a group.

If the table does not belong to the group, raise a 404 error.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset.	required
table	str	Table name.	required
group	SchemaGroup	Group.	required

Source code in pixano/app/routers/utils.py

def assert_table_in_group(dataset: Dataset, table: str, group: SchemaGroup) -> None:
    """Assert that a table belongs to a group.

    If the table does not belong to the group, raise a 404 error.

    Args:
        dataset: Dataset.
        table: Table name.
        group: Group.
    """
    if table in [SchemaGroup.ITEM.value, SchemaGroup.SOURCE.value]:
        return
    elif table not in dataset.schema.groups[group]:
        raise HTTPException(
            status_code=404,
            detail=f"Table {table} is not in the {group.value} group table.",
        )

create_row_handler(dataset_id, group, table, id, row, settings) async

Add a row to a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name.	required
id	str	ID of the row.	required
row	BaseSchemaModel	Row to add.	required
settings	Settings	App settings.	required

Returns:

Type	Description
BaseSchemaModel	The added row.

Source code in pixano/app/routers/utils.py

async def create_row_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    id: str,
    row: BaseSchemaModel,
    settings: Settings,
) -> BaseSchemaModel:
    """Add a row to a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name.
        id: ID of the row.
        row: Row to add.
        settings: App settings.

    Returns:
        The added row.
    """
    if id != row.id:
        raise HTTPException(status_code=400, detail="ID in path and body do not match.")
    return (await create_rows_handler(dataset_id=dataset_id, group=group, table=table, rows=[row], settings=settings))[
        0
    ]

create_rows(dataset, table, models)

Add rows to a table.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset containing the table.	required
table	str	Table name.	required
models	list[BaseSchemaModel]	Models of the rows to add.	required

Returns:

Type	Description
list[BaseSchema]	The added rows.

Source code in pixano/app/routers/utils.py

def create_rows(
    dataset: Dataset,
    table: str,
    models: list[BaseSchemaModel],
) -> list[BaseSchema]:
    """Add rows to a table.

    Args:
        dataset: Dataset containing the table.
        table: Table name.
        models: Models of the rows to add.

    Returns:
        The added rows.
    """
    try:
        schema: type[BaseSchema] = dataset.schema.schemas[table] if table != SchemaGroup.SOURCE.value else Source
        rows: list[BaseSchema] = BaseSchemaModel.to_rows(models, schema)
    except Exception as err:
        raise HTTPException(
            status_code=400,
            detail="Invalid data.\n" + str(err),
        )

    try:
        created_rows = dataset.add_data(table, rows)
    except DatasetIntegrityError as err:
        raise HTTPException(status_code=400, detail="Dataset integrity compromised.\n" + str(err))
    except ValueError as err:
        raise HTTPException(
            status_code=400,
            detail="Invalid data.\n" + str(err),
        )

    return created_rows

create_rows_handler(dataset_id, group, table, rows, settings) async

Add rows to a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name containing the rows.	required
rows	list[BaseSchemaModel]	Rows to add.	required
settings	Settings	App settings.	required

Returns:

Type	Description
list[BaseSchemaModel]	List of updated rows.

Source code in pixano/app/routers/utils.py

async def create_rows_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    rows: list[BaseSchemaModel],
    settings: Settings,
) -> list[BaseSchemaModel]:
    """Add rows to a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name containing the rows.
        rows: Rows to add.
        settings: App settings.

    Returns:
        List of updated rows.
    """
    group = validate_group(group)
    dataset = get_dataset(dataset_id, settings.library_dir, settings.media_dir)
    assert_table_in_group(dataset, table, group)
    rows_rows = create_rows(dataset, table, rows)
    rows_models = get_models_from_rows(table, _SCHEMA_GROUP_TO_SCHEMA_MODEL_DICT[group], rows_rows)
    return rows_models

delete_row_handler(dataset_id, group, table, id, settings) async

Delete a row from a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name containing the row.	required
id	str	ID of the row to delete.	required
settings	Settings	App settings.	required

Source code in pixano/app/routers/utils.py

async def delete_row_handler(dataset_id: str, group: SchemaGroup, table: str, id: str, settings: Settings) -> None:
    """Delete a row from a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name containing the row.
        id: ID of the row to delete.
        settings: App settings.
    """
    return await delete_rows_handler(dataset_id, group, table, ids=[id], settings=settings)

delete_rows(dataset, table, ids)

Delete rows from a table.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset containing the table.	required
table	str	Table name.	required
ids	list[str]	IDs of the rows to delete.	required

Returns:

Type	Description
list[str]	IDs not found.

Source code in pixano/app/routers/utils.py

def delete_rows(
    dataset: Dataset,
    table: str,
    ids: list[str],
) -> list[str]:
    """Delete rows from a table.

    Args:
        dataset: Dataset containing the table.
        table: Table name.
        ids: IDs of the rows to delete.

    Returns:
        IDs not found.
    """
    try:
        ids_not_found = dataset.delete_data(table, ids)
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail="Invalid query parameters. ids and item_ids cannot be set at the same time",
        )
    return ids_not_found

delete_rows_handler(dataset_id, group, table, ids, settings) async

Delete rows from a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name containing the rows.	required
ids	list[str]	IDs of the rows to delete.	required
settings	Settings	App settings.	required

Source code in pixano/app/routers/utils.py

async def delete_rows_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    ids: list[str],
    settings: Settings,
) -> None:
    """Delete rows from a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name containing the rows.
        ids: IDs of the rows to delete.
        settings: App settings.
    """
    group = validate_group(group)
    dataset = get_dataset(dataset_id, settings.library_dir, settings.media_dir)
    assert_table_in_group(dataset, table, group)
    delete_rows(dataset, table, ids)
    return None

get_dataset(dataset_id, dir, media_dir=None)

Get a dataset.

If the dataset is not found, raise a 404 error.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID.	required
dir	Path	Directory containing the dataset.	required
media_dir	Path | None	Directory containing the media files.	None

Returns:

Type	Description
Dataset	The dataset.

Source code in pixano/app/routers/utils.py

def get_dataset(dataset_id: str, dir: Path, media_dir: Path | None = None) -> Dataset:
    """Get a dataset.

    If the dataset is not found, raise a 404 error.

    Args:
        dataset_id: Dataset ID.
        dir: Directory containing the dataset.
        media_dir: Directory containing the media files.

    Returns:
        The dataset.
    """
    try:
        dataset = Dataset.find(dataset_id, dir, media_dir)
    except FileNotFoundError:
        raise HTTPException(
            status_code=404,
            detail=f"Dataset {dataset_id} not found in {dir.absolute()}.",
        )
    return dataset

get_model_from_row(table, model_type, row)

Get a model from a row.

Parameters:

Name	Type	Description	Default
table	str	Table name containing the row.	required
model_type	type[T]	Model type to create.	required
row	BaseSchema	Row.	required

Returns:

Type	Description
T	The model.

Source code in pixano/app/routers/utils.py

def get_model_from_row(table: str, model_type: type[T], row: BaseSchema) -> T:
    """Get a model from a row.

    Args:
        table: Table name containing the row.
        model_type: Model type to create.
        row: Row.

    Returns:
        The model.
    """
    try:
        is_group = issubclass(model_type, BaseSchemaModel)
        if not is_group:
            raise HTTPException(status_code=500, detail="Model type is not a subclass of BaseModelSchema.")
    except TypeError:
        raise HTTPException(status_code=500, detail="Model type is not a subclass of BaseModelSchema.")
    if issubclass(model_type, AnnotationModel):
        group = SchemaGroup.ANNOTATION
    elif issubclass(model_type, EmbeddingModel):
        group = SchemaGroup.EMBEDDING
    elif issubclass(model_type, EntityModel):
        group = SchemaGroup.ENTITY
    elif issubclass(model_type, ItemModel):
        group = SchemaGroup.ITEM
    elif issubclass(model_type, SourceModel):
        group = SchemaGroup.SOURCE
    elif issubclass(model_type, ViewModel):
        group = SchemaGroup.VIEW
    else:
        raise HTTPException(
            status_code=500,
            detail="Model type not correct.",
        )

    pixano_schema_type = get_super_type_from_dict(type(row), _PIXANO_SCHEMA_REGISTRY)

    if pixano_schema_type is None:
        raise HTTPException(
            status_code=500,
            detail="Schema type not found in registry.",
        )
    table_info = TableInfo(name=table, group=group.value, base_schema=pixano_schema_type.__name__)
    model = model_type.from_row(row, table_info)
    return model

get_models_from_rows(table, model_type, rows)

Get models from rows.

Parameters:

Name	Type	Description	Default
table	str	Table name containing the rows.	required
model_type	type[T]	Model type to create.	required
rows	list[BaseSchema]	Rows.	required

Returns:

Type	Description
list[T]	List of models.

Source code in pixano/app/routers/utils.py

def get_models_from_rows(
    table: str,
    model_type: type[T],
    rows: list[BaseSchema],
) -> list[T]:
    """Get models from rows.

    Args:
        table: Table name containing the rows.
        model_type: Model type to create.
        rows: Rows.

    Returns:
        List of models.
    """
    return [get_model_from_row(table, model_type, row) for row in rows]

get_row_handler(dataset_id, group, table, id, settings) async

Get a row model.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name.	required
id	str	ID of the row.	required
settings	Settings	App settings.	required

Returns:

Type	Description
BaseSchemaModel	The model.

Source code in pixano/app/routers/utils.py

async def get_row_handler(
    dataset_id: str, group: SchemaGroup, table: str, id: str, settings: Settings
) -> BaseSchemaModel:
    """Get a row model.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name.
        id: ID of the row.
        settings: App settings.

    Returns:
        The model.
    """
    return (await get_rows_handler(dataset_id, group, table, settings, ids=[id], item_ids=None, limit=None, skip=0))[0]

get_rows(dataset, table, where=None, ids=None, item_ids=None, limit=None, skip=0)

Get rows from a table.

The rows can be filtered by a where clause, IDs, item IDs or a limit and a skip.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset containing the table.	required
table	str	Table name.	required
where	str | None	Where clause.	None
ids	list[str] | None	IDs of the rows.	None
item_ids	list[str] | None	Item IDs of the rows.	None
limit	int | None	Limit number of rows.	None
skip	int	Skip number of rows.	0

Returns:

Type	Description
list[BaseSchema]	List of rows.

Source code in pixano/app/routers/utils.py

def get_rows(
    dataset: Dataset,
    table: str,
    where: str | None = None,
    ids: list[str] | None = None,
    item_ids: list[str] | None = None,
    limit: int | None = None,
    skip: int = 0,
) -> list[BaseSchema]:
    """Get rows from a table.

    The rows can be filtered by a where clause, IDs, item IDs or a limit and a skip.

    Args:
        dataset: Dataset containing the table.
        table: Table name.
        where: Where clause.
        ids: IDs of the rows.
        item_ids: Item IDs of the rows.
        limit: Limit number of rows.
        skip: Skip number of rows.

    Returns:
        List of rows.
    """
    try:
        rows = dataset.get_data(table_name=table, where=where, ids=ids, limit=limit, skip=skip, item_ids=item_ids)
    except DatasetPaginationError as err:
        raise HTTPException(status_code=400, detail="Invalid query parameters. " + str(err))
    except DatasetAccessError as err:
        raise HTTPException(status_code=500, detail="Internal server error. " + str(err))

    if rows == [] or rows is None:
        raise HTTPException(
            status_code=404,
            detail=f"No rows found for {dataset.info.id}/{table}.",
        )
    return rows

get_rows_handler(dataset_id, group, table, settings, where=None, ids=None, item_ids=None, limit=None, skip=0) async

Get row models.

Rows can be filtered by a where clause, IDs, item IDs or a limit and a skip.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name.	required
settings	Settings	App settings.	required
where	str | None	Where clause.	None
ids	list[str] | None	IDs of the rows.	None
item_ids	list[str] | None	Item IDs of the rows.	None
limit	int | None	Limit number of rows.	None
skip	int	Skip number of rows.	0

Returns:

Type	Description
list[BaseSchemaModel]	List of models.

Source code in pixano/app/routers/utils.py

async def get_rows_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    settings: Settings,
    where: str | None = None,
    ids: list[str] | None = None,
    item_ids: list[str] | None = None,
    limit: int | None = None,
    skip: int = 0,
) -> list[BaseSchemaModel]:
    """Get row models.

    Rows can be filtered by a where clause, IDs, item IDs or a limit and a skip.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name.
        settings: App settings.
        where: Where clause.
        ids: IDs of the rows.
        item_ids: Item IDs of the rows.
        limit: Limit number of rows.
        skip: Skip number of rows.

    Returns:
        List of models.
    """
    group = validate_group(group)
    dataset = get_dataset(dataset_id, settings.library_dir, settings.media_dir)
    assert_table_in_group(dataset, table, group)
    rows = get_rows(dataset=dataset, table=table, where=where, ids=ids, item_ids=item_ids, limit=limit, skip=skip)
    models = get_models_from_rows(table, _SCHEMA_GROUP_TO_SCHEMA_MODEL_DICT[group], rows)
    return models

update_row_handler(dataset_id, group, table, id, row, settings) async

Update a row in a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name containing the row.	required
id	str	ID of the row.	required
row	BaseSchemaModel	Row to update.	required
settings	Settings	App settings.	required

Returns:

Type	Description
BaseSchemaModel	The updated row.

Source code in pixano/app/routers/utils.py

async def update_row_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    id: str,
    row: BaseSchemaModel,
    settings: Settings,
) -> BaseSchemaModel:
    """Update a row in a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name containing the row.
        id: ID of the row.
        row: Row to update.
        settings: App settings.

    Returns:
        The updated row.
    """
    if id != row.id:
        raise HTTPException(status_code=400, detail="ID in path and body do not match.")
    return (await update_rows_handler(dataset_id=dataset_id, group=group, table=table, rows=[row], settings=settings))[
        0
    ]

update_rows(dataset, table, models)

Update rows in a table.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset containing the table.	required
table	str	Table name.	required
models	list[BaseSchemaModel]	Models of the rows to update.	required

Returns:

Type	Description
list[BaseSchema]	The updated rows.

Source code in pixano/app/routers/utils.py

def update_rows(
    dataset: Dataset,
    table: str,
    models: list[BaseSchemaModel],
) -> list[BaseSchema]:
    """Update rows in a table.

    Args:
        dataset: Dataset containing the table.
        table: Table name.
        models: Models of the rows to update.

    Returns:
        The updated rows.
    """
    try:
        schema: type[BaseSchema] = dataset.schema.schemas[table] if table != SchemaGroup.SOURCE.value else Source
        rows: list[BaseSchema] = BaseSchemaModel.to_rows(models, schema)
    except Exception as err:
        raise HTTPException(
            status_code=400,
            detail="Invalid data.\n" + str(err),
        )

    try:
        updated_rows = dataset.update_data(table, rows)
    except DatasetIntegrityError as err:
        raise HTTPException(status_code=400, detail="Dataset integrity compromised.\n" + str(err))
    except ValueError as err:
        raise HTTPException(
            status_code=400,
            detail="Invalid data.\n" + str(err),
        )

    # TODO: return updated rows instead of input rows
    # TODO: check if rows are updated or created which change HTTP status code
    return updated_rows

update_rows_handler(dataset_id, group, table, rows, settings) async

Update rows in a table.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
group	SchemaGroup	Schema group of the schema of the table.	required
table	str	Table name containing the rows.	required
rows	list[BaseSchemaModel]	Rows to update.	required
settings	Settings	App settings.	required

Returns:

Type	Description
list[BaseSchemaModel]	List of updated rows.

Source code in pixano/app/routers/utils.py

async def update_rows_handler(
    dataset_id: str,
    group: SchemaGroup,
    table: str,
    rows: list[BaseSchemaModel],
    settings: Settings,
) -> list[BaseSchemaModel]:
    """Update rows in a table.

    Args:
        dataset_id: Dataset ID containing the table.
        group: Schema group of the schema of the table.
        table: Table name containing the rows.
        rows: Rows to update.
        settings: App settings.

    Returns:
        List of updated rows.
    """
    group = validate_group(group)
    dataset = get_dataset(dataset_id, settings.library_dir, settings.media_dir)
    assert_table_in_group(dataset, table, group)
    row_rows = update_rows(dataset, table, rows)
    row_models = get_models_from_rows(table, _SCHEMA_GROUP_TO_SCHEMA_MODEL_DICT[group], row_rows)
    return row_models

validate_group(group, valid_groups=set(SchemaGroup))

Assert that a group is valid.

Parameters:

Name	Type	Description	Default
group	str | SchemaGroup	Group.	required
valid_groups	set[SchemaGroup]	The valid groups.	set(SchemaGroup)

Source code in pixano/app/routers/utils.py

def validate_group(
    group: str | SchemaGroup,
    valid_groups: set[SchemaGroup] = set(SchemaGroup),
) -> SchemaGroup:
    """Assert that a group is valid.

    Args:
        group: Group.
        valid_groups: The valid groups.
    """
    try:
        group = SchemaGroup(group)
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail=f"Group {group} is not a SchemaGroup.",
        )
    if group not in valid_groups:
        raise HTTPException(
            status_code=400,
            detail=f"Group {group.value} is not valid.",
        )
    return group