Flows
Access flows through client.flows in the Datature Vi SDK.
Flows are training workflow configurations. Each flow defines how a model will be trained: dataset selection, model architecture, hyperparameters, and training schedule. When you execute a flow, it creates a run. You can create, update, list, and delete flows through the SDK.
Before You Start
- Vi SDK installed with authentication configured
- A training project with workflows created
- Familiarity with training workflows and runs
Methods
list()
List all training workflows in your organization.
flows = client.flows.list()
for flow in flows.items:
print(f"Flow: {flow.flow_id}")
print(f"Created: {flow.metadata.time_created}")for page in client.flows.list():
for flow in page.items:
print(f"Flow: {flow.flow_id}")
print(f" Name: {flow.spec.name}")
print(f" Created: {flow.metadata.time_created}")
print(f" Blocks: {len(flow.spec.blocks)}")from datetime import datetime, timedelta
cutoff = (datetime.now() - timedelta(days=7)).timestamp() * 1000
recent_flows = []
for page in client.flows.list():
for flow in page.items:
if flow.metadata.time_created > cutoff:
recent_flows.append(flow)
print(f"Found {len(recent_flows)} recent flows")all_flows = list(client.flows.list().all_items())
print(f"Total flows: {len(all_flows)}")Returns: PaginatedResponse[Flow]
get()
Get a specific training flow by ID.
flow = client.flows.get("flow_abc123")
print(f"Flow ID: {flow.flow_id}")
print(f"Name: {flow.spec.name}")
print(f"Blocks: {len(flow.spec.blocks)}")flow = client.flows.get("flow_abc123")
flow.info() # Prints formatted flow summaryflow = client.flows.get("flow_abc123")
print(f"Flow: {flow.spec.name}")
print(f"Blocks ({len(flow.spec.blocks)}):")
for i, block in enumerate(flow.spec.blocks, 1):
print(f" {i}. {block.block}")
if block.settings:
for key, value in list(block.settings.items())[:3]:
print(f" {key}: {value}")flow = client.flows.get("flow_abc123")
print("Global Settings:")
for key, value in flow.spec.settings.items():
print(f" {key}: {value}")
print("\nTolerations:")
for key, values in flow.spec.tolerations.items():
print(f" {key}: {values}")flow = client.flows.get("flow_abc123")
print(f"Organization: {flow.organization_id}")
print(f"Schema: {flow.spec.schema}")
print(f"ETag: {flow.etag}")
if flow.spec.training_project:
print(f"Training Project: {flow.spec.training_project}")Returns: Flow
delete()
Delete a training flow.
deleted = client.flows.delete("flow_abc123")flow = client.flows.get("flow_abc123")
print(f"About to delete flow: {flow.spec.name}")
print(f" Blocks: {len(flow.spec.blocks)}")
confirm = input("Delete? (yes/no): ")
if confirm.lower() == "yes":
client.flows.delete("flow_abc123")
print("Deleted.")Returns: DeletedFlow
create()
Create a new training flow.
flow = client.flows.create(
spec={
"name": "My Training Flow",
"schema": "v1",
"blocks": [],
"settings": {},
"tolerations": {}
}
)
print(f"Created flow: {flow.flow_id}")
print(f"Name: {flow.spec.name}")flow = client.flows.create(
spec={
"name": "Production Flow",
"schema": "v1",
"blocks": [
{
"block": "data_loader",
"settings": {"batch_size": 32},
"style": {}
}
],
"settings": {"learning_rate": 0.001},
"tolerations": {}
},
metadata={"environment": "production", "version": "1.0"}
)Returns: Flow
update()
Update an existing training flow. Uses PATCH semantics: only provided fields are modified.
flow = client.flows.update(
flow_id="flow_abc123",
spec={"name": "Updated Flow Name"}
)
print(f"Updated: {flow.spec.name}")flow = client.flows.update(
flow_id="flow_abc123",
spec={"training_project": "project_xyz789"},
metadata={"status": "active"}
)Returns: Flow
Response types
Flow
from vi.api.resources.flows.responses import FlowMethods: info() → prints a formatted flow summary.
FlowSpec
from vi.api.resources.flows.responses import FlowSpecFlowBlock
from vi.api.resources.flows.responses import FlowBlockRelated resources
Updated 30 days ago