Skip to main content

Agent SDK

Classes for Agent-triggered flows

Objects for triggering a Flow from an Agent can be found in agent_sdk for Agents v5.0+.

FileWatcherResult Class

FileWatcherResult is a dictionary of FileParam objects indexed by node name.param name.

  • param files: dict[str, fileParam] - Dictionary of FileParam objects indexed by node name.param name
  • param tags: list[FileTag] | None - List of tags to be applied to all files

TriggerFlowParams Class

TriggerFlowParams specifies the inputs for the Flow executed when all files are observed. It includes the following parameters:

  • param single_file_params: dict[str, FileParam] | None - Dict of FileParam objects indexed by node name.param name. These parameters are used for Nodes that accept a single file as input.
  • param multi_file_params: dict[str, MultiFileParam] | None - Dict of MultiFileParam objects indexed by node name.param name. These parameters are used for Nodes that accept multiple files as input.
  • param benchling_tag: Tag | None - Additional parameters to be passed to flow. This parameter is used for inputs to the Input_Benchling node.
  • param additional_params: dict[str, str] | None - Additional parameters to be passed to flow. This parameter is used for inputs to the Input_Param node; the key is the name if the Node name for the input parameter, and the value is the string to pass into the Node.

FileParam Class

FileParam specifies files to be uploaded to Ganymede Cloud and their corresponding Flow parameters. These parameters are provided to the execute function once all files are detected.

  • param filename: str - Name of the file, e.g. "my_file.txt"
  • param content_type: str - Content type of the file, e.g. "text/plain". If not specified, the content type of the first file in the files dict will be used.
  • param body: bytes - File contents in bytes
  • param param: str - Name of parameter to be used in Flow, e.g. node_name.parameter_field_name
  • param parent_dir: str - Path within the Agent watch directory containing the file. For example. if C:/Users/username/watch_dir/ is being watched and C:/Users/username/watch_dir/abc/def/my_file.txt is found, then parent_dir would be "abc/def"
  • param upload_ts: str - Timestamp string in ISO format of when file was uploaded to the Agent watch directory, e.g. "2021-01-01T00:00:00Z"
  • param upload_path: str | None - Path in Ganymede storage where file will be uploaded
  • param tags: list[FileTag] | None - List of tags to be applied to the file
  • param bucket_name: str - Bucket associated with file
  • param files: str - Alternative method for specifying file contents, where the key is the filename and the value is the file body.

MultiFileParam Class

MultiFileParam is used for submitting multiple files to a single node. It includes the following parameters:

  • param files: str - Alternative method for specifying file contents, where the key is the filename and the value is the file body.
  • param content_type: str - Content type of file, e.g. "text/plain". If not specified, the content type of the first file in the files dict will be used.
  • param param: str - Name of parameter to be used in flow, e.g. node_name.parameter_field_name
  • param parent_dir: str - Path within Agent watch directory that contains file. For example. if C:/Users/username/watch_dir/ is being watched and C:/Users/username/watch_dir/abc/def/my_file.txt is found, then parent_dir would be "abc/def"
  • param upload_ts: str - Timestamp string in ISO format of when file was uploaded to Agent watch directory, e.g. "2021-01-01T00:00:00Z"
  • param upload_paths: list[str] | None - Path in Ganymede storage where file will be uploaded
  • param tags: list[FileTag] | None - List of tags to be applied to file
  • param bucket_name: str - Bucket associated with file

The MultiFileParam object contains a method for initiation from a list of FileParam objects as shown below. The content type of the object is assumed to take on the content type of the first item in the list.

# assume fp1 and fp2 are FileParam objects
m = agent_sdk.file_params_list_to_multi([fp1, fp2])

Utility functions

Agent utility functions are provided in agent_sdk for validating data integrity and interacting with file systems.

note

The agent_sdk is only available for Agents v5.0+. Prior to v5.0, these functions were included in ganymede_sdk.agent.

Computing file checksums

Ganymede provides functions to validate file integrity; these values can be used to verify the integrity of a file uploaded to cloud storage:

# Before Agent v5.0
# from ganymede_sdk.agent.utils import calculate_md5, calculate_crc32c

from agent_sdk import calculate_md5, calculate_crc32c

file_path = "path/to/local/file"

# either md5 or crc32c can be used to validate the integrity of a file
md5 = calculate_md5(file_path)
crc32c = calculate_crc32c(file_path)

You can also calculate the checksum of a file uploaded to Ganymede Cloud by creating a a tempfile.TemporaryFile object, writing the file contents to it, and then calculating the checksum:

from agent_sdk import calculate_md5, calculate_crc32c
import os
import tempfile

data = b"Example data to calculate checksum"

with tempfile.NamedTemporaryFile(delete=False) as tmp_file:
tmp_file.write(data)
tmp_file_name = tmp_file.name

md5 = calculate_md5(tmp_file_name)
crc32c = calculate_crc32c(tmp_file_name)

os.remove(tmp_file_name)

File system utilities

agent_sdk provides a number of convenience functions, which can be helpful to use with cron Agents that involve more complex logic prior to invoking a flow. Some examples of this are when a file is written to multiple times before being processed, or if there is a variable number of files being processed, such that the trigger for invoking a flow requires more than just the presence of a file.

ScanResult Dataclass

ScanResult stores file paths for files of interest. It includes:

  • param file_path: str - Path to file
  • param modified_time: datetime - Datetime of when file was last modified

Functions

list_files_recursive returns a list of all files in a directory and its subdirectories.

  • param file_path: str - Path to directory to list files from

matches_pattern returns True if a file path matches at least one of the specified regex patterns specified and False otherwise.

  • param filename: str - Name of file
  • param pattern: str | re.Pattern - Regex pattern or list of regex patterns to match against

is_file_ready returns True if a file has the modified time is within the last interval_in_seconds seconds, or if the size of the file has changed in that same timespan.

  • param file_path: str - Path to file to watch
  • param threshold_seconds: int - Number of seconds to wait between checks, by default 0.1

get_most_recent_access_result returns a ScanResult object referencing the most recently accessed file in a directory. Access time is updated when a file is read from or written to.

  • param directory: str - Path to directory to watch

filter_by_age returns a list of files that have not been modified within the last age_in_minutes minutes.

  • param scan_results: list[ScanResult] - List of ScanResult objects
  • param age_in_minutes: int - Minimum age in minutes

zip_directory creates a zip file of a directory and its contents.

  • param directory: str - Path to directory to zip
  • param zip_file: str - Path to zip file to create

scan_for_finished_files scans a directory, returning paths to files with a modified date older than the specified number of minutes

  • param directory: str - Path to directory to scan
  • param age_in_minutes: int - Minimum age in minutes for files to be included in the results
  • param pattern: re.Pattern | list[re.Pattern] - Regex pattern to match files against; only files that match against at least one of the specified patterns will be included in results

Example Use Case

You can use scan_for_finished_files to continuously scan a directory for files, uploading them to Ganymede Cloud for processing when they are older than a specified number of minutes. The Flow could query previously uploaded files using the list_files method to avoid uploading the same file multiple times.

Querying Ganymede from Agent Code

from agent_sdk.query import read_sql_query

df = read_sql_query('SELECT * FROM instrument_methods')

Logging Methods

Ganymede Agents (v4.9+) support user-defined logging messages in the agent_sdk, aligning with logging level for Agent messages. Each level corresponds with a separate method in agent_sdk.

from agent_sdk import internal, debug, info, activity, error

# log internal
internal('Display internal message')

# log debug
debug('Display debug message')

# log info
info('Display info message')

# log activity
activity('Display activity message')

# log error
error('Display error message')

In the UI, these log messages are viewable and filterable on the corresponding Connections page.