result = await provider.deactivate_partner(partner_id)
if result.is_ok:
print("Partner deactivated successfully")
else:
print(f"Error: {result.error.code} - {result.error.message}")
result = await provider.validate_partner_signature(
consumer_key=consumer_key,
signature=signature,
payload=payload,
timestamp=timestamp,
)
if not result.is_ok:
print(f"Error: {result.error.code}") # PARTNER_INACTIVE
Error: partner_inactive
result = await provider.reactivate_partner(partner_id)
if result.is_ok:
print("Partner reactivated — can sign requests again")
# All active partners
result = await provider.list_partners(is_active=True)
if result.is_ok:
partners = result.unwrap()
print(f"Active partners: {len(partners)}")
for p in partners:
print(f" {p.id}: {p.partner_name} (active: {p.is_active})")
# Filter by organization
result = await provider.list_partners(
organization_id="default",
is_active=None, # All statuses
)
if result.is_ok:
partners = result.unwrap()
print(f"\nAll partners in 'default': {len(partners)}")
# All partners (no filters)
result = await provider.list_partners()
Partner deactivated successfully
Error: partner_inactive
Partner reactivated — can sign requests again
Active partners: 3
partner-001: Acme Corp (active: True)
partner-002: Beta Inc (active: True)
partner-003: Gamma Ltd (active: True)
All partners in 'default': 4
"""GL IAM Manage Partner Lifecycle Example."""
import asyncio
import hashlib
import hmac
from datetime import datetime, timezone
from gl_iam.core.crypto_config import CryptoConfig, EncryptionAlgorithm
from gl_iam.core.types.sso import SSOPartnerCreate
from gl_iam.providers.postgresql import (
PostgreSQLPartnerRegistryProvider,
PostgreSQLConfig,
)
DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/gliam"
SECRET_KEY = "your-secret-key-min-32-characters-long" # For JWT signing
# Generate with: python -c "import secrets, base64; print(base64.urlsafe_b64encode(secrets.token_bytes(32)).decode())"
ENCRYPTION_KEY = "<YOUR_ENCRYPTION_KEY>"
async def main():
# Setup
config = PostgreSQLConfig(
database_url=DATABASE_URL,
secret_key=SECRET_KEY,
crypto_config=CryptoConfig(
encryption_keys={1: ENCRYPTION_KEY},
encryption_algorithm=EncryptionAlgorithm.AES_256_GCM,
),
)
provider = PostgreSQLPartnerRegistryProvider(config)
# Register a partner
reg_result = await provider.register_partner(
SSOPartnerCreate(partner_name="Lifecycle Demo", org_id="default")
)
registration = reg_result.unwrap()
partner_id = registration.partner.id
consumer_key = registration.consumer_key
consumer_secret = registration.consumer_secret
print(f"1. Registered: {registration.partner.partner_name}")
# Helper to compute and validate signature
async def try_signature() -> bool:
payload = '{"test": true}'
timestamp = datetime.now(timezone.utc).isoformat()
signing_string = f"{timestamp}|{consumer_key}|{payload}"
signature = hmac.new(
consumer_secret.encode(), signing_string.encode(), hashlib.sha256,
).hexdigest()
result = await provider.validate_partner_signature(
consumer_key=consumer_key, signature=signature,
payload=payload, timestamp=timestamp,
)
return result.is_ok
# Verify active partner can sign
print(f"2. Signature valid (active): {await try_signature()}")
# Deactivate
deactivate_result = await provider.deactivate_partner(partner_id)
print(f"3. Deactivated: {deactivate_result.is_ok}")
# Verify deactivated partner is rejected
print(f"4. Signature valid (inactive): {await try_signature()}")
# Reactivate
reactivate_result = await provider.reactivate_partner(partner_id)
print(f"5. Reactivated: {reactivate_result.is_ok}")
# Verify reactivated partner can sign again
print(f"6. Signature valid (reactivated): {await try_signature()}")
# List all partners
list_result = await provider.list_partners(organization_id="default")
if list_result.is_ok:
partners = list_result.unwrap()
print(f"\n7. Partners in 'default': {len(partners)}")
for p in partners:
print(f" {p.id}: {p.partner_name} (active: {p.is_active})")
# List only active partners
active_result = await provider.list_partners(is_active=True)
if active_result.is_ok:
print(f"\n8. Active partners: {len(active_result.unwrap())}")
await provider.close()
if __name__ == "__main__":
asyncio.run(main())