Quickstart¶
This guide will get you up and running with the Tessera SDK in 5 minutes.
Prerequisites¶
- A running Tessera server (see Tessera docs)
- Python 3.10+
- The SDK installed:
pip install tessera-sdk
Basic Usage¶
Create a client¶
from tessera_sdk import TesseraClient
# Connect to local server
client = TesseraClient(base_url="http://localhost:8000")
# Or with authentication
client = TesseraClient(
base_url="http://localhost:8000",
api_key="your-api-key"
)
Create a team¶
Create an asset¶
asset = client.assets.create(
fqn="warehouse.analytics.dim_customers",
owner_team_id=team.id,
description="Customer dimension table"
)
print(f"Created asset: {asset.id}")
Publish a contract¶
result = client.assets.publish_contract(
asset_id=asset.id,
schema={
"type": "object",
"properties": {
"customer_id": {"type": "integer"},
"email": {"type": "string"},
"created_at": {"type": "string", "format": "date-time"}
},
"required": ["customer_id", "email"]
},
version="1.0.0"
)
if result.published:
print(f"Published contract v{result.contract.version}")
else:
print(f"Breaking change detected, proposal created: {result.proposal.id}")
Check impact before changes¶
impact = client.assets.check_impact(
asset_id=asset.id,
proposed_schema={
"type": "object",
"properties": {
"customer_id": {"type": "string"}, # Type changed!
"email": {"type": "string"}
}
}
)
if impact.safe_to_publish:
print("Safe to publish - no breaking changes")
else:
print(f"Breaking changes: {impact.breaking_changes}")
print(f"Affected consumers: {len(impact.affected_registrations)}")
Register as a consumer¶
registration = client.registrations.create(
contract_id=contract.id,
consumer_team_id=my_team.id
)
print(f"Registered for contract: {registration.id}")
Complete Example¶
from tessera_sdk import TesseraClient
def main():
client = TesseraClient(base_url="http://localhost:8000")
# Setup
team = client.teams.create(name="analytics-team")
asset = client.assets.create(
fqn="analytics.core.users",
owner_team_id=team.id
)
# Publish initial contract
result = client.assets.publish_contract(
asset_id=asset.id,
schema={
"type": "object",
"properties": {
"user_id": {"type": "integer"},
"email": {"type": "string"}
},
"required": ["user_id"]
},
version="1.0.0"
)
print(f"Published: {result.contract.version}")
# Check impact of adding a required field
impact = client.assets.check_impact(
asset_id=asset.id,
proposed_schema={
"type": "object",
"properties": {
"user_id": {"type": "integer"},
"email": {"type": "string"},
"name": {"type": "string"} # New field
},
"required": ["user_id", "name"] # name is now required!
}
)
print(f"Safe to publish: {impact.safe_to_publish}")
print(f"Breaking changes: {impact.breaking_changes}")
if __name__ == "__main__":
main()
Next steps¶
- Configuration - Advanced configuration options
- Async Usage - Use the async client
- Error Handling - Handle errors gracefully