bartender.staging#

Attributes#

CHUNK_SIZE

Exceptions#

UnsupportedContentTypeError

When content type is unsupported.

Functions#

has_needed_files(→ Literal[True])

Check if files required by application are present in job directory.

create_job_dir(→ pathlib.Path)

Create job directory.

_is_valid_content_type(→ bool)

unpack_upload(→ None)

Unpack archive file to job id directory.

Module Contents#

bartender.staging.CHUNK_SIZE#
exception bartender.staging.UnsupportedContentTypeError#

Bases: Exception

When content type is unsupported.

bartender.staging.has_needed_files(application: bartender.config.ApplicatonConfiguration, job_dir: pathlib.Path) → Literal[True]#

Check if files required by application are present in job directory.

Parameters:
Raises:

IndexError – When one or more needed files can not be found

Returns:

True when found or no files where needed.

Return type:

Literal[True]

async bartender.staging.create_job_dir(job_id: int, job_root_dir: pathlib.Path) → pathlib.Path#

Create job directory.

Parameters:

  • job_id (int) – id of the job.

  • job_root_dir (pathlib.Path) – Root directory for all jobs.

Raises:

HTTPException – When job directory could not be made.

Returns:

Directory of job.

Return type:

pathlib.Path

bartender.staging._is_valid_content_type(content_type: str | None) → bool#
Parameters:

content_type (Optional[str]) –

Return type:

bool

async bartender.staging.unpack_upload(job_dir: pathlib.Path, archive: fastapi.UploadFile, dest_fn: str = 'archive.zip') → None#

Unpack archive file to job id directory.

Parameters:

  • job_dir (pathlib.Path) – Where to put archive file.

  • archive (fastapi.UploadFile) – The archive file with async read method.

  • dest_fn (str) – Temporary archive filename.

Return type:

None