CHAP.foxden package

Submodules

CHAP.foxden.models module

FOXDEN Pydantic model classes.

class FoxdenRequestConfig(*, did: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None)] | None = None, limit: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None = 10, query: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None)] | None = None, url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None)], verbose: bool | None = None)

Bases: CHAPBaseModel

FOXDEN HTTP request base configuration class.

Parameters:

• did (string, optional) – FOXDEN dataset identifier (did).

• limit (int, optional) – Maximum number of returned records, defaults to 10.

• query (string, optional) – FOXDEN query.

• verbose (bool, optional) – Verbose output flag, defaults to False.

did: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None)] | None

limit: Annotated[int, None, Interval(gt=0, ge=None, lt=None, le=None), None] | None

model_config: ClassVar[ConfigDict] = {}

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

query: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=True, strict=True, min_length=None, max_length=None, pattern=None)] | None

url: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=True, min_length=None, max_length=None, pattern=None)]

verbose: bool | None

CHAP.foxden.processor module

File : processor.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: Processor module for FOXDEN services

class FoxdenProvenanceProcessor(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: Processor

Processor to collect CHAP workflow provenance data.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

process(data)

Extract provenance data from the pipeline data for submission to the FOXDEN Provenance service.

Parameters:

data (list[PipelineData]) – Input data.

Returns:

CHAP workflow provenance record.

Return type:

dict

CHAP.foxden.reader module

File : reader.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: FOXDEN readers

class FoxdenDataDiscoveryReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: PipelineItem

Reader for the FOXDEN Data Discovery service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

read(config)

Read records from the FOXDEN Data Discovery service based on did or an arbitrary query.

Parameters:

config (CHAP.foxden.models.FoxdenRequestConfig) – FOXDEN HTTP request configuration.

Returns:

Discovered data records.

Return type:

list

class FoxdenMetadataReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: PipelineItem

Reader for the FOXDEN Metadata service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

read(config)

Read records from the FOXDEN Metadata service based on did or an arbitrary query.

Parameters:

config (CHAP.foxden.models.FoxdenRequestConfig) – FOXDEN HTTP request configuration.

Returns:

Metadata records.

Return type:

list

class FoxdenProvenanceReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: PipelineItem

Reader for FOXDEN Provenance data from a specific FOXDEN Provenance service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

read(config)

Read records from the FOXDEN Provenance service based on did or an arbitrary query.

Parameters:

config (CHAP.foxden.models.FoxdenRequestConfig) – FOXDEN HTTP request configuration.

Returns:

Provenance input and output file records.

Return type:

list

class FoxdenSpecScansReader(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None)

Bases: PipelineItem

Reader for FOXDEN SpecScans data from a specific FOXDEN SpecScans service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

read(url, data, did='', query='', spec=None, method='POST', verbose=False)

Read and return data from a specific FOXDEN SpecScans service.

Parameters:

• url (str) – URL of service.

• data (list[PipelineData]) – Input data.

• did (string, optional) – FOXDEN dataset identifier (did).

• query (string, optional) – FOXDEN query.

• spec (dictionary, optional) – FOXDEN spec.

• method (str, optional) – HTTP method to use, ‘POST’ for creation or ‘PUT’ for update, defaults to ‘POST’.

• verbose (bool, optional) – Verbose output flag, defaults to False.

Returns:

Contents of the input data.

Return type:

object

CHAP.foxden.utils module

FOXDEN utils module.

HttpRequest(url, payload, method='POST', headers=None, scope='read', timeout=10, dry_run=False)

Submit a HTTP request to a FOXDEN service

Parameters:

• url (str) – URL of service.

• payload (str) – HTTP request payload.

• method (str, optional) – HTTP method to use, defaults to ‘POST’.

• headers (dictionary, optional) – HTTP headers to use.

• scope (string, optional) – FOXDEN scope: ‘read’ or ‘write’, defaults to ‘read’.

• timeout (str, optional) – Timeout of HTTP request, defaults to 10.

• dry_run (bool, optional) – dry_run option to verify HTTP workflow, defaults to False.

Returns:

HTTP response.

Return type:

requests.models.Response

readFoxdenToken(scope)

Obtain a FOXDEN token.

Parameters:

scope (string) – FOXDEN scope: ‘read’ or ‘write’.

CHAP.foxden.writer module

File : writer.py Author : Valentin Kuznetsov <vkuznet AT gmail dot com> Description: FOXDEN writers

class FoxdenDoiWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for saving info to the FOXDEN DOI service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

write(url, data, provider='Datacite', description='', draft=True, publishMetadata=True, verbose=False)

Write data to the FOXDEN DOI service.

Parameters:

• data (list[PipelineData]) – Input data.

• url (str) – URL of service.

• provider (str, optional) – DOI provider name, e.g. Zenodo, Datacite, Materialcommons, defaults to ‘Datacite’.

• description (str, optional) – Dataset description.

• draft (bool, optional) – Draft DOI flag, defaults to True.

• publishMetadata (bool, optional) – Publish metadata with DOI, defaults to True.

• verbose (bool, optional) – Verbose output flag, defaults to False.

Returns:

HTTP response from FOXDEN DOI service.

Return type:

list[dict]

class FoxdenMetadataWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for saving data to the FOXDEN Metadata service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

write(data, url)

Write data to the FOXDEN Metadata service.

Parameters:

• data (list[PipelineData]) – Input data.

• config (CHAP.foxden.models.FoxdenRequestConfig) – FOXDEN HTTP request configuration.

Returns:

HTTP response from FOXDEN Metadata service.

Return type:

list[dict]

class FoxdenProvenanceWriter(*, root: Annotated[Path, PathType(path_type=dir)] | None = '/home/runner/work/ChessAnalysisPipeline/ChessAnalysisPipeline/docs', inputdir: Annotated[Path, PathType(path_type=dir)] | None = None, outputdir: Annotated[Path, PathType(path_type=dir)] | None = None, interactive: bool | None = False, log_level: Literal['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'] | None = 'INFO', logger: Logger | None = None, name: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, schema: Annotated[str, StringConstraints(strip_whitespace=True, to_upper=None, to_lower=None, strict=None, min_length=1, max_length=None, pattern=None)] | None = None, filename: str, force_overwrite: bool | None = False, remove: bool | None = False)

Bases: Writer

Writer for saving data to the FOXDEN Provenance service.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

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

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Args:

self: The BaseModel instance. context: The context.

write(data, url)

Write data to the FOXDEN Provenance service.

Parameters:

• data (list[PipelineData]) – Input data.

• config (CHAP.foxden.models.FoxdenRequestConfig) – FOXDEN HTTP request configuration.

Returns:

HTTP response from FOXDEN Provenance service.

Return type:

list[dict]

Module contents

This subpackage contains PipelineItems to communicate with FOXDEN services.