pixano.app.routers.sources

create_source(dataset_id, id, source, settings) async

Add a source in the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
id	str	ID of the source.	required
source	SourceModel	Source to add.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

Returns:

Type	Description
SourceModel	The source added.

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

@router.post("/{dataset_id}/{id}", response_model=SourceModel)
async def create_source(
    dataset_id: str,
    id: str,
    source: SourceModel,
    settings: Annotated[Settings, Depends(get_settings)],
) -> SourceModel:
    """Add a source in the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        id: ID of the source.
        source: Source to add.
        settings: App settings.

    Returns:
        The source added.
    """
    return await create_row_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, id, source, settings)

create_sources(dataset_id, sources, settings) async

Add sources in the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
sources	list[SourceModel]	Sources to add.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

Returns:

Type	Description
list[SourceModel]	List of sources added.

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

@router.post("/{dataset_id}/", response_model=list[SourceModel])
async def create_sources(
    dataset_id: str,
    sources: list[SourceModel],
    settings: Annotated[Settings, Depends(get_settings)],
) -> list[SourceModel]:
    """Add sources in the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        sources: Sources to add.
        settings: App settings.

    Returns:
        List of sources added.
    """
    return await create_rows_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, sources, settings)

delete_source(dataset_id, id, settings) async

Delete a source from the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
id	str	ID of the source to delete.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

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

@router.delete("/{dataset_id}/{id}")
async def delete_source(dataset_id: str, id: str, settings: Annotated[Settings, Depends(get_settings)]) -> None:
    """Delete a source from the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        id: ID of the source to delete.
        settings: App settings.
    """
    return await delete_row_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, id, settings)

delete_sources(dataset_id, ids, settings) async

Delete sources from the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
ids	Annotated[list[str], Query()]	IDs of the sources to delete.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

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

@router.delete("/{dataset_id}/")
async def delete_sources(
    dataset_id: str,
    ids: Annotated[list[str], Query()],
    settings: Annotated[Settings, Depends(get_settings)],
) -> None:
    """Delete sources from the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        ids: IDs of the sources to delete.
        settings: App settings.
    """
    return await delete_rows_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, ids, settings)

get_source(dataset_id, id, settings) async

Get a source from the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
id	str	ID of the source.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

Returns:

Type	Description
SourceModel	The source.

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

@router.get("/{dataset_id}/{id}", response_model=SourceModel)
async def get_source(dataset_id: str, id: str, settings: Annotated[Settings, Depends(get_settings)]) -> SourceModel:
    """Get a source from the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        id: ID of the source.
        settings: App settings.

    Returns:
        The source.
    """
    return await get_row_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, id, settings)

get_sources(dataset_id, settings, ids=Query(None), limit=None, skip=0, where=None) async

Get sources from the 'source' table of a dataset.

They can be filtered by IDs, a where clause or paginated.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required
ids	list[str] | None	IDs of the sources.	Query(None)
limit	int | None	Limit number of sources.	None
skip	int	Skip number of sources.	0
where	str | None	Where clause.	None

Returns:

Type	Description
list[SourceModel]	List of sources.

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

@router.get("/{dataset_id}/", response_model=list[SourceModel])
async def get_sources(
    dataset_id: str,
    settings: Annotated[Settings, Depends(get_settings)],
    ids: list[str] | None = Query(None),
    limit: int | None = None,
    skip: int = 0,
    where: str | None = None,
) -> list[SourceModel]:
    """Get sources from the `'source'` table of a dataset.

    They can be filtered by IDs, a where clause or paginated.

    Args:
        dataset_id: Dataset ID containing the table.
        settings: App settings.
        ids: IDs of the sources.
        limit: Limit number of sources.
        skip: Skip number of sources.
        where: Where clause.

    Returns:
        List of sources.
    """
    return await get_rows_handler(
        dataset_id=dataset_id,
        group=SchemaGroup.SOURCE,
        table=SchemaGroup.SOURCE.value,
        settings=settings,
        where=where,
        ids=ids,
        item_ids=None,
        limit=limit,
        skip=skip,
    )

update_source(dataset_id, id, source, settings) async

Update a source in the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
id	str	ID of the source.	required
source	SourceModel	Source to update.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

Returns:

Type	Description
SourceModel	The source updated.

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

@router.put("/{dataset_id}/{id}", response_model=SourceModel)
async def update_source(
    dataset_id: str,
    id: str,
    source: SourceModel,
    settings: Annotated[Settings, Depends(get_settings)],
) -> SourceModel:
    """Update a source in the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        id: ID of the source.
        source: Source to update.
        settings: App settings.

    Returns:
        The source updated.
    """
    return await update_row_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, id, source, settings)

update_sources(dataset_id, sources, settings) async

Update sources in the 'source' table of a dataset.

Parameters:

Name	Type	Description	Default
dataset_id	str	Dataset ID containing the table.	required
sources	list[SourceModel]	Sources to update.	required
settings	Annotated[Settings, Depends(get_settings)]	App settings.	required

Returns:

Type	Description
list[SourceModel]	List of sources updated.

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

@router.put("/{dataset_id}/", response_model=list[SourceModel])
async def update_sources(
    dataset_id: str,
    sources: list[SourceModel],
    settings: Annotated[Settings, Depends(get_settings)],
) -> list[SourceModel]:
    """Update sources in the `'source'` table of a dataset.

    Args:
        dataset_id: Dataset ID containing the table.
        sources: Sources to update.
        settings: App settings.

    Returns:
        List of sources updated.
    """
    return await update_rows_handler(dataset_id, SchemaGroup.SOURCE, SchemaGroup.SOURCE.value, sources, settings)