File System Connector
File System Connectors give workflows programmatic access to directories on lab instruments. Using the UniteLabs SDK, you can list files, stream real-time change notifications, retrieve file metadata, and upload files to Object Storage — all without touching the instrument OS directly.
Prerequisites
- A running File System Connector visible in the Connectors view of the platform.
- A UniteLabs account with SDK access.
unitelabs-sdkinstalled in your Python environment.
Connect to a connector
from unitelabs.sdk import AsyncApiClient
async with AsyncApiClient() as client:
device = await client.get_service_by_name("File System (LabChip)")
List available folders
Connectors expose only pre-approved directories to prevent arbitrary filesystem access. Use get_folder_whitelist() to retrieve the allowed set:
whitelist = await device.folder_service.get_folder_whitelist()
print(whitelist)
# ['/data/labchip/results', '/data/labchip/archive']
List files in a folder
files = await device.folder_service.get_folder_contents(
path="/data/labchip/results"
)
for f in files:
print(f)
# /data/labchip/results/run_001_PeakTable.csv
# /data/labchip/results/run_001_SizeTable.csv
# /data/labchip/results/run_001_WellTable.csv
Get file metadata
Retrieve size, owner, and timestamps before deciding whether to process a file:
meta = await device.file_service.get_file_metadata(
path="/data/labchip/results/run_001_PeakTable.csv"
)
print(meta["Metadata"]["Size"]["value"]) # file size in bytes
print(meta["Metadata"]["LastModified"]["value"]) # ISO timestamp
print(meta["Metadata"]["Owner"]["value"]) # OS user ("unknown" on Windows)
Watch for new files
subscribe_changes() is the key real-time capability. It returns an async context manager that yields the path of each new or modified file as it appears:
async with AsyncApiClient() as client:
device = await client.get_service_by_name("File System (LabChip)")
async with await device.folder_service.subscribe_changes() as subscription:
async for changed_path in subscription:
print(f"New or modified file: {changed_path}")
# trigger processing here
subscribe_changes() is a long-running async generator. Run it inside a Prefect @task or a dedicated async loop — do not call it in a blocking synchronous context.Upload a file to Object Storage
Transfer a file from the instrument directory directly to S3 — no local intermediate copy:
s3_url = (
f"s3s://{access_key}:{secret_key}"
f"@{host}:{port}/{bucket}/data/instruments/labchip/run_001_PeakTable.csv"
)
await device.s3_service.upload_file(
path="/data/labchip/results/run_001_PeakTable.csv",
s3_url=s3_url,
)
Construct the S3 URL dynamically from values loaded out of the datalake-admin secret. See Object Storage for the URL format and credential loading pattern.
Putting it together
A complete async function that combines folder listing, change watching, and S3 upload:
from unitelabs.sdk import AsyncApiClient
async def monitor_and_upload(connector_name: str, s3_base_url: str) -> None:
"""Watch a connector for new files and upload each one to Object Storage."""
async with AsyncApiClient() as client:
device = await client.get_service_by_name(connector_name)
whitelist = await device.folder_service.get_folder_whitelist()
print(f"Watching {whitelist[0]} on '{connector_name}'")
async with await device.folder_service.subscribe_changes() as sub:
async for file_path in sub:
meta = await device.file_service.get_file_metadata(path=file_path)
size = meta["Metadata"]["Size"]["value"]
print(f"Detected: {file_path} ({size} bytes)")
filename = file_path.split("/")[-1]
s3_url = f"{s3_base_url}/{filename}"
await device.s3_service.upload_file(path=file_path, s3_url=s3_url)
print(f"Uploaded to: {s3_url}")
Next steps
- Object Storage — understand the S3 URL scheme, bucket layout, and credentials
- Building an ETL — embed this pattern in a production Prefect flow with startup reconciliation and Warehouse writes