Models#

Job requests#

class JobRequest(name, description, template_file, scheduled_execution_date=None)#

Provides the abstract base class representing a job request.

Each subclass represents a specific job type and may override some steps of the submission process. They also add additional file types and properties as required.

Parameters:

namestr: Name of the job as shown in the job queue.
descriptionOptional[str]: Description of the job as shown in the job queue.
template_filestr or pathlib.Path, default: None: Template to use the job.
scheduled_execution_datedatetime.datetime, default: None: Earliest date and time to run the job. If no date and time are provided, the job begins as soon as possible.

class ExcelExportJobRequest(name, description, template_file, database_key, records, scheduled_execution_date=None)#

Represents an Excel export job request.

An Excel template and references to the records to export are required.

Subclass of JobRequest.

Parameters:

namestr: Name of the job as shown in the job queue.
descriptionOptional[str]: Description of the job as shown in the job queue.
template_filestr or pathlib.Path: Excel template file.
database_keystr: Database key for the records to export.
recordslist of ExportRecord: List of objects representing the records to export.
scheduled_execution_datedatetime.datetime, default: None: Earliest date and time to run the job. If no date and time are provided, the job begins as soon as possible.

Examples

>>> template_file: pathlib.Path  # pathlib Path object for the template
>>> record_history_identities = [12345, 23456]
>>> job_request = ExcelExportJobRequest(
...     name="Excel export job",
...     description=None,
...     template_file=template_file,
...     database_key="MI_Training",
...     records=[ExportRecord(rhid) for rhid in record_history_identities],
... )
>>> job_request
<ExcelExportJobRequest: name: "Excel export job">

>>> tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
>>> job_request = ExcelExportJobRequest(
...     name="Excel export job (future execution)",
...     description="Example job request to run in the future",
...     template_file=template_file,
...     database_key="MI_Training",
...     records=[ExportRecord(rhid) for rhid in record_history_identities],
...     scheduled_execution_date=tomorrow,
... )
>>> job_request
<ExcelExportJobRequest: name: "Excel export job (future execution)">

class ExcelImportJobRequest(name, description, template_file=None, data_files=None, combined_files=None, attachment_files=None, scheduled_execution_date=None)#

Represents an Excel import job request.

This class supports either combined imports (with a template and data in the same file) or separate data and template imports.

Subclass of JobRequest.

Parameters:

namestr: Name of the job as shown in the job queue.
descriptionOptional[str]: Description of the job as shown in the job queue.
template_filestr, pathlib.Path, or JobFile, default: None: Excel template file.
data_fileslist of str or pathlib.Path or JobFile, default: None: Excel files containing the data to import.
combined_fileslist of str or pathlib.Path or JobFile, default: None: Excel files containing data and template information.
attachment_fileslist of str or pathlib.Path or JobFile, default: None: Any other files referenced in the data or combined files.
scheduled_execution_datedatetime.datetime, default: None: Earliest date and time to run the job. If no date and time are provided, the job begins as soon as possible.

Notes

Changed in version 1.1: Path parameters now accept JobFile inputs.

Examples

>>> template_file: pathlib.Path  # pathlib Path object for the template
>>> job_request = ExcelImportJobRequest(
...     name="Excel import job",
...     description=None,
...     data_files=["assets/data_file_1.xlsx", "assets/data_file_2.xlsx"],
...     template_file=template_file,
... )
>>> job_request
<ExcelImportJobRequest: name: "Excel import job">

>>> tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
>>> job_request = ExcelImportJobRequest(
...     name="Excel import job (future execution)",
...     description="Example job request to run in the future",
...     data_files=["assets/data_file_1.xlsx", "assets/data_file_2.xlsx"],
...     template_file=template_file,
...     scheduled_execution_date=tomorrow,
... )
>>> job_request
<ExcelImportJobRequest: name: "Excel import job (future execution)">

class TextImportJobRequest(name, description, template_file=None, data_files=None, attachment_files=None, scheduled_execution_date=None)#

Represents a text import job request.

This class requires a template file and one or more data files.

Subclass of JobRequest.

Parameters:

namestr: Name of the job as shown in the job queue.
descriptionOptional[str]: Description of the job as shown in the job queue.
template_filestr or pathlib.Path or JobFile, default: None: Text import template file.
data_fileslist of str or pathlib.Path or JobFile, default: None: Text files containing the data to import.
attachment_fileslist of str or pathlib.Path or JobFile, default: None: Any other files referenced in the data files.
scheduled_execution_datedatetime.datetime, default: None: Earliest date and time to run the job. If no date and time are provided, the job begins as soon as possible.

Notes

Changed in version 1.1: Path parameters now accept JobFile inputs.

Examples

>>> template_file: pathlib.Path  # pathlib Path object for the template
>>> job_request = TextImportJobRequest(
...     name="Text import job",
...     description=None,
...     template_file=template_file,
...     data_files=["Data_File_1.txt", "Data_File_2.txt"],  # Relative paths
... )
>>> job_request
<TextImportJobRequest: name: "Text import job">

>>> tomorrow = datetime.datetime.now(datetime.UTC) + datetime.timedelta(days=1)
>>> job_request = TextImportJobRequest(
...     name="Text import job (future execution)",
...     description="Example job request to run in the future",
...     template_file=template_file,
...     data_files=["Data_File_1.txt", "Data_File_2.txt"],
...     scheduled_execution_date=tomorrow,
... )
>>> job_request
<TextImportJobRequest: name: "Text import job (future execution)">

Jobs#

class AsyncJob(job_obj, job_queue_api)#

Base class that represents a job on the server.

This class provides information on the current status of the job and any job-specific outputs. It allows modification of job metadata, such as the job name, description, and scheduled execution date.

classmethod create_job(job_obj, job_queue_api)#

Create an instance of a JobQueue AsyncJob subclass.

Returns:

AsyncJob: The appropriate AsyncJob subclass based on the response from the server.

Notes

Note

Do not use this method to create an AsyncJob object. The create_job() and create_job_and_wait() methods should be used instead, which will return correctly-instantiated AsyncJob objects.

property id: str#

Unique job ID, which is the tecommended way to refer to individual jobs.

Returns:

str: Unique ID of the job.

property name: str#

Display name of the job, which does not have to be unique.

Returns:

str: Display name of the job.

update_name(value)#

Update the display name of the job on the server.

This method performs an HTTP request against the Granta MI Server API.

Parameters:

valuestr: New name for the job.

Raises:

ValueError: If the job has been deleted from the server.

property description: str | None#

Description of the job as shown in Granta MI.

Returns:

str or None: Description of the job.

update_description(value)#

Update the job description on the server.

This method performs an HTTP request against the Granta MI Server API.

Parameters:

valuestr: New description for the job.

Raises:

ValueError: If the job has been deleted from the server.

property status: JobStatus#

Job status of the job on the server.

Returns:

JobStatus: Status of the job.

Notes

Note

A return value of JobStatus.Succeeded does not mean that the import or export operation itself was successful, it only means that the job was successfully attempted. For more detailed information on the job status, check the contents of the AsyncJob.output_information property.

property type: JobType#

Type of the job on the server.

Returns:

JobType: Type of the job.

property position: int | None#

Position of the job in the job queue.

Returns:

int or None: Position of the job in the job queue or None if the job is not currently pending.

move_to_top()#

Promote the job to the top of the job queue.

To use this method, you must have MI_ADMIN permission.

property submitter_information: Dict[str, str | datetime | List[str]]#

Information about the job submission.

Returns:

dict: Dictionary of job submission information with the username of the submitter, date and time of submission, and the roles that the submitter belongs to (indexed by name).

property completion_date_time: datetime | None#

Date and time of job completion.

Returns:

datetime.datetime or None: Date and time of job completion or None if the job is pending.

property execution_date_time: datetime | None#

Date and time that the job was run.

Returns:

datetime.datetime or None: Date and time that the job was run or None if the job is pending.

property scheduled_execution_date_time: datetime | None#

Date and time that the job is scheduled to run.

Returns:

datetime.datetime or None: Date and time that the job is scheduled to run or None if the job is not scheduled.

update_scheduled_execution_date_time(value)#

Update the date and time that the job is scheduled to run on the server.

Performs an HTTP request against the Granta MI Server API.

Parameters:

valuedatetime.datetime: New date and time that the job is scheduled to run.

Raises:

ValueError: If the job has been deleted from the server.

property output_information: Dict[str, Any] | None#

Additional output information provided by the job (if supported by the job type).

Additional output information typically includes record placement, a summary of the success of the job, and more verbose logging. The additional information supported is dependent on the job.

Returns:

dict or None: Additional output information provided by the job.

property output_file_names: List[str] | None#

List of names of the job’s output files.

Returns:

list of str or None: List of the job’s output files or None if the job has no output files.

download_file(remote_file_name, file_path)#

Download an output file from the server by name and save it to a specified location.

Performs an HTTP request against the Granta MI Server API.

Parameters:

remote_file_namestr: Filename provided by the output_file_names() method.
file_pathstr or pathlib.Path: Path to save the file to.

Raises:

KeyError: If the filename does not exist for this job.
ValueError: If the job has been deleted from the server.

Examples

>>> job: AsyncJob
>>> folder = pathlib.Path(r"C:\path\to\folder")  # or Linux equivalent
>>> for file_name in job.output_file_names:
...     job.download_file(file_name, folder / file_name)
>>> print(list(folder.iterdir()))
[Path(C:/path/to/folder/output_1.json), Path(C:/path/to/folder/output_...

get_file_content(remote_file_name)#

Download an output file from the server by name and return the file contents.

Performs an HTTP request against the Granta MI Server API.

Parameters:

remote_file_namestr: Filename provided by the output_file_names() method.

Returns:

bytes: Content of the specified file.

Raises:

KeyError: If the filename does not exist for this job.
ValueError: If the job has been deleted from the server.

Examples

>>> job: AsyncJob
>>> file_content = {}
>>> for file_name in job.output_file_names:
...     file_content[file_name] = job.get_file_content(file_name)
>>> print(file_content)
{'output_1.log': b'2024-03-11 17:24:16,342 [396] INFO  Task started:...

update()#

Update the job from the server.

Raises:

ValueError: If the job has been deleted from the server.

class ImportJob(job_obj, job_queue_api)#

Represents an import job on the server. Subclass of AsyncJob.

Objects of this type are returned from the create_job() and create_job_and_wait() methods after submitting a ExcelImportJobRequest or TextImportJobRequest to the server.

Notes

Note

Do not instantiate this class directly.

Added in version 1.0.1.

property status: JobStatus#

Job status of the job on the server.

Returns:

JobStatus: Status of the job.

Notes

Note

A return value of JobStatus.Succeeded does not mean that the import or export operation itself was successful, it only means that the job was successfully attempted. For more detailed information on the job status, check the contents of the AsyncJob.output_information property.

class ExportJob(job_obj, job_queue_api)#

Represents an export job on the server. Subclass of AsyncJob.

Objects of this type are returned from the create_job() and create_job_and_wait() methods after submitting a ExcelExportJobRequest to the server.

Notes

Note

Do not instantiate this class directly.

Added in version 1.0.1.

Other models#

class JobFile(path, virtual_path)#

Represents a file associated with a JobRequest.

JobFile can be used instead of paths in import requests. This allows building a virtual file structure for the import, to work around some path limitations and makes working with attachments easier.

Added in version 1.1.

Parameters:

pathstr or pathlib.Path: Path to the local file to use in a JobRequest. Can be absolute or relative.
virtual_pathstr or pathlib.Path: Virtual path to use to refer to the file in the JobRequest. Must be relative.

Examples

Using an import template from a shared network location and placing it at the root of the isolated job environment.

>>> template_file = JobFile(pathlib.Path(r"\\server\share\Tensile_Import_Template_v1.xslx"), "template.xslx")

JobFiles can be used to set up a file structure that matches the expectation of the import, without moving or copying files to the client machine or requiring the script to run in a specific location. For example, if the Excel or text import template is configured to import a picture to a Picture attribute, the data file includes a reference to the picture by relative path, e.g. /assets/panel_front.jpg.

>>> data_file = JobFile(pathlib.Path(r"C:\test_results\tensile\sample_001\data.xslx"), "data.xslx")
>>> attachment_file = JobFile(pathlib.Path(r"C:\test_results\pictures\sample_001\panel_front.png"), "./assets/panel_front.png")

class ExportRecord(record_history_identity, record_version=None)#

Defines a record to include in an export job.

Parameters:

record_history_identityint: History identities of a record to export. You can find history identities in the Granta MI Viewer or using the Scripting Toolkit.
record_versionint, default: None: Specific version of the record to export. If no version is specified for version-controlled records, the latest available version for the current user is exported.

class JobQueueProcessingConfiguration(purge_job_age_in_milliseconds, purge_interval_in_milliseconds, polling_interval_in_milliseconds, concurrency)#

Provides a read-only configuration of the job queue on the server.

Parameters:

purge_job_age_in_millisecondsint: Age at which to automatically remove jobs from the queue.
purge_interval_in_millisecondsint: Time between purge operations.
polling_interval_in_millisecondsint: Idle time before executing the next job in the queue.
concurrencyint: Maximum number of jobs to process concurrently.

enum JobStatus(value)#

Provides possible states of a job in the job queue.

Valid values are as follows:

Pending = <JobStatus.Pending: 'Pending'>#: Job is in the queue.

Running = <JobStatus.Running: 'Running'>#: Job is currently executing.

Succeeded = <JobStatus.Succeeded: 'Succeeded'>#: Job has completed (does not guarantee that no errors occurred).

Failed = <JobStatus.Failed: 'Failed'>#: Job could not complete.

Cancelled = <JobStatus.Cancelled: 'Cancelled'>#: Job was cancelled by the user.

Deleted = <JobStatus.Deleted: 'Deleted'>#: Job was deleted on the server.

enum JobType(value)#

Provides possible job types.

Valid values are as follows:

ExcelImportJob = <JobType.ExcelImportJob: 'ExcelImportJob'>#

ExcelExportJob = <JobType.ExcelExportJob: 'ExcelExportJob'>#

TextImportJob = <JobType.TextImportJob: 'TextImportJob'>#