Managing secrets via Python SDK
What is the Python SDK
The Python SDK is a library for programmatic interaction with Passwork. It's designed for advanced automation where CLI capabilities aren't enough.
| Feature | Description |
|---|---|
| Reading secrets | Retrieve password field values, custom fields, and attachments by ID or search |
| Modifying secrets | Update password field values, custom fields, tags, and descriptions |
| Creating secrets | Programmatically add records to specific folders |
| Managing structure | Work with folders and vaults, move records around |
When to use SDK vs CLI
| Scenario | Recommendation |
|---|---|
| CI/CD pipeline, deploy script | passwork-cli — simpler, no code needed |
| Password rotation with custom logic | Python SDK — more flexibility, better error handling |
| Migrating from another system | Python SDK — transform data as needed |
| Checking secret integrity | Python SDK — implement complex validation |
| Grabbing a single secret in bash | passwork-cli get — just one command |
Installation
Install with pip from PyPI or GitHub:
# PyPI
pip install passwork-python
# Or from GitHub via SSH
pip install git+ssh://[email protected]:passwork-me/passwork-python.git
# Or from GitHub via HTTPS
pip install git+https://github.com/passwork-me/passwork-python.git
See also: Python connector.
Initializing the client
from passwork_client import PassworkClient
client = PassworkClient(host="https://passwork.example.com")
client.set_tokens("your_access_token", "your_refresh_token") # refresh_token is optional
client.set_master_key("your_master_key") # if client-side encryption is enabled
tip
Avoid hardcoding tokens in your code. Use environment variables instead:
import os
from passwork_client import PassworkClient
client = PassworkClient(host=os.environ["PASSWORK_HOST"])
client.set_tokens(
os.environ["PASSWORK_ACCESS_TOKEN"],
os.environ.get("PASSWORK_REFRESH_TOKEN"),
)
if os.environ.get("PASSWORK_MASTER_KEY"):
client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])
Reading secrets
By record ID
item = client.get_item("<item-id>")
# Standard fields
print(item["login"])
print(item["password"])
# Custom fields (customs)
for custom in item.get("customs", []):
if custom["name"] == "DB_HOST":
db_host = custom["value"]
if custom["name"] == "API_KEY":
api_key = custom["value"]
By folder
# Fetch all records from a folder
items = client.search_and_decrypt(folder_ids=["<folder-id>"])
for item in items:
print(f"{item['name']}: {item['login']}")
By search string
# Find records matching a query
items = client.search_and_decrypt(query="app-db prod")
for item in items:
print(f"{item['id']}: {item['name']}")
Modifying secrets
Updating an existing record
# Load the record
item = client.get_item("<item-id>")
# Prepare update payload
updated_data = {
"vaultId": item["vaultId"],
"password": "new-strong-password",
"customs": [
{"name": "API_KEY", "value": "new-api-key", "type": "password"},
# keep remaining customs that should be preserved
*[c for c in item.get("customs", []) if c["name"] != "API_KEY"],
],
}
# Persist changes
client.update_item(item["id"], updated_data)
Creating a new record
item_id = client.create_item({
"vaultId": "<vault-id>",
"folderId": "<folder-id>", # optional
"name": "Production DB",
"login": "app_user",
"password": "secure-password",
"customs": [
{"name": "DB_HOST", "value": "db.example.com", "type": "text"},
{"name": "DB_PORT", "value": "5432", "type": "text"},
],
"tags": ["prod", "database"],
})
Bulk operations
The SDK enables automation such as:
- moving records between folders;
- adding or removing tags in bulk;
- updating fields based on a template;
- migrating the
secrets/*hierarchy.
# Add a tag to all records in a folder
items = client.search_and_decrypt(folder_ids=["<folder-id>"])
for item in items:
tags = list(item.get("tags") or [])
if "legacy" not in tags:
tags.append("legacy")
client.update_item(item["id"], {
"vaultId": item["vaultId"],
"tags": tags,
})
Practical examples
Rotating a database password
Full example with error handling for PostgreSQL password rotation:
import os
import secrets
import psycopg2
from passwork_client import PassworkClient
def get_custom(item: dict, name: str, default=None):
for custom in item.get("customs", []):
if custom["name"] == name:
return custom["value"]
return default
def rotate_db_password(item_id: str, db_role: str):
"""Rotate the password for a PostgreSQL role."""
client = PassworkClient(host=os.environ["PASSWORK_HOST"])
client.set_tokens(
os.environ["PASSWORK_ACCESS_TOKEN"],
os.environ.get("PASSWORK_REFRESH_TOKEN"),
)
if os.environ.get("PASSWORK_MASTER_KEY"):
client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])
# Create a fresh password
new_password = secrets.token_urlsafe(32)
# Retrieve admin credentials for the DB connection
admin_item = client.get_item(os.environ["DB_ADMIN_ITEM_ID"])
try:
# Apply the new password in PostgreSQL
conn = psycopg2.connect(
host=get_custom(admin_item, "DB_HOST"),
dbname="postgres",
user=admin_item["login"],
password=admin_item["password"],
)
conn.autocommit = True
with conn.cursor() as cur:
# Use a parameterized query for safety
cur.execute(
"ALTER ROLE %s WITH PASSWORD %s",
(db_role, new_password)
)
conn.close()
# Store the updated password in Passwork
item = client.get_item(item_id)
client.update_item(item_id, {
"vaultId": item["vaultId"],
"password": new_password,
})
print(f"Password rotated for {db_role}")
return True
except Exception as e:
print(f"Rotation failed: {e}")
return False
# Usage
rotate_db_password(
item_id="<item-id>",
db_role="app_user"
)
Checking secret validity
Script that periodically verifies stored credentials still work:
import os
import psycopg2
from passwork_client import PassworkClient
def get_custom(item: dict, name: str, default=None):
for custom in item.get("customs", []):
if custom["name"] == name:
return custom["value"]
return default
def check_db_credentials(folder_id: str):
"""Verify that database credentials in Passwork are valid."""
client = PassworkClient(host=os.environ["PASSWORK_HOST"])
client.set_tokens(
os.environ["PASSWORK_ACCESS_TOKEN"],
os.environ.get("PASSWORK_REFRESH_TOKEN"),
)
if os.environ.get("PASSWORK_MASTER_KEY"):
client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])
items = client.search_and_decrypt(folder_ids=[folder_id])
broken = []
for item in items:
if "database" not in (item.get("tags") or []):
continue
try:
conn = psycopg2.connect(
host=get_custom(item, "DB_HOST"),
dbname=get_custom(item, "DB_NAME", "postgres"),
user=item["login"],
password=item["password"],
connect_timeout=5,
)
conn.close()
print(f"✅ {item['name']}")
except Exception as e:
print(f"❌ {item['name']}: {e}")
broken.append(item)
# Flag the record for review
tags = list(item.get("tags") or [])
if "needs-review" not in tags:
tags.append("needs-review")
client.update_item(item["id"], {
"vaultId": item["vaultId"],
"tags": tags,
})
return broken
# Usage
broken = check_db_credentials(folder_id="<folder-id>")
if broken:
print(f"\n{len(broken)} secrets need review")
Importing secrets from files
Migrate secrets from .env files into Passwork:
import os
from pathlib import Path
from passwork_client import PassworkClient
def migrate_env_file(env_path: str, vault_id: str, folder_id: str | None, tags: list):
"""Import secrets from a .env file into Passwork."""
client = PassworkClient(host=os.environ["PASSWORK_HOST"])
client.set_tokens(
os.environ["PASSWORK_ACCESS_TOKEN"],
os.environ.get("PASSWORK_REFRESH_TOKEN"),
)
if os.environ.get("PASSWORK_MASTER_KEY"):
client.set_master_key(os.environ["PASSWORK_MASTER_KEY"])
env_file = Path(env_path)
secrets_dict = {}
# Parse the .env file
for line in env_file.read_text().splitlines():
line = line.strip()
if not line or line.startswith("#"):
continue
if "=" in line:
key, value = line.split("=", 1)
secrets_dict[key.strip()] = value.strip().strip('"\'')
# Store in Passwork
item_data = {
"vaultId": vault_id,
"name": env_file.stem,
"customs": [
{"name": key, "value": value, "type": "text"}
for key, value in secrets_dict.items()
],
"tags": tags,
}
if folder_id:
item_data["folderId"] = folder_id
client.create_item(item_data)
print(f"Imported {len(secrets_dict)} secrets from {env_path}")
# Usage
migrate_env_file(
env_path="./legacy/.env.production",
vault_id="<vault-id>",
folder_id="<folder-id>",
tags=["prod", "migrated"],
)