| Name | identityservice-pythonsdk JSON |
| Version |
1.0.8
JSON |
| download |
| home_page | None |
| Summary | Python SDK for IdentityService integration with multi-tenant authentication, JWT management, and gRPC services |
| upload_time | 2025-09-13 01:18:02 |
| maintainer | None |
| docs_url | None |
| author | D3 Security |
| requires_python | >=3.8 |
| license | MIT License
Copyright (c) 2025 D3 Security
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE. |
| keywords |
authentication
jwt
etcd
identity
security
multi-tenant
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# IdentityService Python SDK
Python SDK for IdentityService integration, providing multi-tenant authentication, JWT token management, and distributed configuration management.
## Features
- **Ed25519 JWT Authentication**: Modern cryptographic signatures for enhanced security
- **Multi-tenant Support**: Tenant-specific configuration and key management
- **Real-time Configuration**: Live configuration updates via etcd watching
- **Service Registration**: Automatic service discovery and health monitoring
- **High-performance Caching**: Memory and Redis-based caching with TTL
- **Framework Integration**: FastAPI and gRPC middleware for seamless integration
- **gRPC TestConnection Service**: Built-in TestConnection service for service health checks
- **Comprehensive Security**: Key rotation, rate limiting, and audit logging
## Installation
```bash
# Install from PyPI
pip install identityservice-pythonsdk==1.0.6
# Or install from source
pip install -e .
```
### Requirements
- Python 3.8+
- etcd cluster (for distributed configuration)
- Redis (optional, for caching)
## Quick Start
### 1. Environment Configuration
Set the following environment variables:
```bash
export D3_TENANT_GUID="your-tenant-guid"
export D3_API_KEY="your-tenant-api-key"
export ETCD_ENDPOINTS="etcd1:2379,etcd2:2379,etcd3:2379"
export SERVICE_NAME="YourService"
```
### 2. Basic Usage
```python
import asyncio
from d3_identity_client import D3IdentityClient
async def main():
# Create and initialize client
async with D3IdentityClient() as client:
# Get tenant information
tenant_info = await client.get_tenant_info()
print(f"Tenant: {tenant_info.tenant_name}")
# Generate service token
token = await client.generate_service_token(
service_name="MyService",
permissions=["read", "write"]
)
# Validate token
result = await client.validate_token(token)
if result.is_valid:
print(f"Token valid for tenant: {result.claims.tenant_name}")
if __name__ == "__main__":
asyncio.run(main())
```
### 3. FastAPI Integration
```python
from fastapi import FastAPI, Depends
from d3_identity_client import D3IdentityClient, create_service_auth_dependency
app = FastAPI()
# Initialize client (do this at startup)
identity_client = None
@app.on_event("startup")
async def startup():
global identity_client
identity_client = await create_client()
@app.on_event("shutdown")
async def shutdown():
if identity_client:
await identity_client.cleanup()
# Create authentication dependency
auth_required = create_service_auth_dependency(
identity_client.auth_service,
required_service="MyService"
)
@app.get("/protected")
async def protected_endpoint(auth_context=auth_required):
return {
"message": "Success",
"tenant": auth_context.tenant_name,
"service": auth_context.get_service_name()
}
```
### 4. gRPC TestConnection Integration
Add TestConnection service to your existing CommandService gRPC server:
```python
import grpc
from concurrent import futures
from d3_identity_client import D3IdentityClient, IdentityServiceOptions
from d3_identity_client.controllers.grpc_controller import (
TestConnectionGrpcServicer,
add_servicer_to_server
)
async def integrate_with_command_service():
# Initialize identity client
config = IdentityServiceOptions(
tenant_guid="your-tenant-guid",
tenant_api_key="your-api-key",
etcd_endpoints=["localhost:2379"],
service_name="CommandService"
)
identity_client = D3IdentityClient(config)
await identity_client.initialize()
# Create your existing gRPC server
server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
# Add your existing services
# your_pb2_grpc.add_YourServiceServicer_to_server(YourServicer(), server)
# Add TestConnection service to the SAME server
test_servicer = TestConnectionGrpcServicer(
identity_client.auth_service,
identity_client.tenant_service
)
add_servicer_to_server(test_servicer, server)
# Start on port 50051 (same port as your existing services)
server.add_insecure_port('[::]:50051')
server.start()
print("✅ CommandService with TestConnection running on port 50051")
return server, identity_client
# Usage
server, client = await integrate_with_command_service()
```
**Call TestConnection from other services:**
```python
import grpc
from protos import test_connection_pb2, test_connection_pb2_grpc
# Connect to CommandService (same port for all services)
channel = grpc.insecure_channel('localhost:50051')
stub = test_connection_pb2_grpc.TestConnectionGrpcServiceStub(channel)
# Call TestConnection with JWT authentication
metadata = [('authorization', 'Bearer your-jwt-token')]
request = test_connection_pb2.TestConnectionRequest()
response = stub.TestConnection(request, metadata=metadata)
print(f"Tenant: {response.TenantName}")
print(f"Status: {response.Status.message}")
```
## Configuration
### Environment Variables
| Variable | Description | Default |
|----------|-------------|---------|
| `D3_TENANT_GUID` | Tenant GUID | Required |
| `D3_API_KEY` | Tenant API key | Required |
| `ETCD_ENDPOINTS` | etcd endpoints (comma-separated) | `localhost:2379` |
| `ETCD_USERNAME` | etcd authentication username | None |
| `ETCD_PASSWORD` | etcd authentication password | None |
| `CACHE_TYPE` | Cache type: `memory`, `redis`, `hybrid` | `memory` |
| `REDIS_HOST` | Redis host for caching | `localhost` |
| `REDIS_PORT` | Redis port | `6379` |
| `SERVICE_NAME` | Service name for registration | None |
| `LOG_LEVEL` | Logging level | `INFO` |
### Programmatic Configuration
```python
from d3_identity_client import IdentityServiceOptions, create_production_config
config = create_production_config(
tenant_guid="your-tenant-guid",
api_key="your-api-key",
service_name="YourService",
etcd_endpoints=["etcd1:2379", "etcd2:2379"],
redis_host="redis.example.com"
)
client = D3IdentityClient(config)
```
## Advanced Usage
### Tenant Watching
Monitor tenant configuration changes in real-time:
```python
async def tenant_change_callback(tenant_info):
print(f"Tenant configuration updated: {tenant_info.tenant_name}")
# Update application configuration based on tenant changes
await client.watch_tenant(tenant_guid, tenant_change_callback)
```
### Service Discovery
Register and discover internal services:
```python
# Register service
await client.register_internal_service(
service_name="MyService",
service_endpoints={
"http": "http://myservice:8080",
"grpc": "grpc://myservice:50051"
}
)
# Discover services
services = await client.get_all_internal_services()
command_service = services.get("CommandService")
if command_service:
http_endpoint = command_service.get_endpoint("http")
```
### Custom Authentication
Implement custom authentication logic:
```python
from d3_identity_client import AuthService
async def custom_auth_handler(token: str):
# Validate token
context = await client.authenticate_token(token)
if not context:
raise AuthenticationError("Invalid token")
# Check custom business rules
if not context.token_claims.has_permission("custom_action"):
raise AuthorizationError("Insufficient permissions")
return context
```
### Multi-tenant Operations
Work with multiple tenants:
```python
# Get multiple tenants
tenant_guids = ["guid1", "guid2", "guid3"]
tenants = await client.get_multiple_tenants(tenant_guids)
for guid, tenant in tenants.items():
if tenant and tenant.is_active_tenant():
print(f"Active tenant: {tenant.tenant_name}")
```
## Security Features
### Key Rotation
The client automatically handles key rotation:
```python
# Keys are automatically rotated by the Identity Service
# Client handles both current and previous keys for validation
# No manual intervention required
```
### Rate Limiting
Configure tenant-specific rate limiting:
```python
tenant_info = await client.get_tenant_info()
api_limit = tenant_info.get_rate_limit_config("api_requests")
if api_limit:
print(f"API rate limit: {api_limit.max_requests_per_minute()}/minute")
```
### Permissions
Check permissions in your application:
```python
async def check_permissions(auth_context, required_permission):
if not await client.auth_service.check_permission(auth_context, required_permission):
raise PermissionError(f"Permission '{required_permission}' required")
```
## Health Monitoring
Monitor client health:
```python
@app.get("/health")
async def health_check():
health_status = await client.health_check()
if health_status["status"] == "healthy":
return {"status": "OK", "details": health_status}
else:
raise HTTPException(status_code=503, detail="Service unhealthy")
```
## Error Handling
The client provides comprehensive error handling:
```python
from d3_identity_client import TokenValidationResult
async def safe_token_validation(token: str):
try:
result = await client.validate_token(token)
if result.is_valid:
return result.claims
else:
logger.warning(f"Token validation failed: {result.error_message}")
return None
except Exception as e:
logger.error(f"Token validation error: {e}")
return None
```
## Development Mode
For development and testing:
```python
from d3_identity_client import create_development_config
config = create_development_config(
tenant_guid="dev-tenant-guid",
api_key="dev-api-key",
service_name="DevService"
)
config.debug_mode = True
config.logging.log_level = LogLevel.DEBUG
client = D3IdentityClient(config)
```
## Testing
Mock the client for unit testing:
```python
import pytest
from unittest.mock import AsyncMock
@pytest.fixture
async def mock_client():
client = AsyncMock(spec=D3IdentityClient)
client.authenticate_token.return_value = AuthenticationContext(...)
return client
```
## Migration from Legacy Authentication
Replace existing JWT authentication:
```python
# Before (legacy)
def verify_jwt_token(token):
return jwt.decode(token, secret_key, algorithms=["HS256"])
# After (D3 Identity Service)
async def verify_jwt_token(token):
result = await client.validate_token(token)
return result.claims if result.is_valid else None
```
## Performance Considerations
- **Caching**: Enable Redis caching for production deployments
- **Connection Pooling**: Client automatically manages etcd connections
- **Token Caching**: JWT validation results are cached for performance
- **Batch Operations**: Use `get_multiple_tenants()` for bulk operations
## Troubleshooting
### Common Issues
1. **etcd Connection Failed**
```python
# Check etcd endpoints and credentials
health = await client.health_check()
print(health["etcd"])
```
2. **Token Validation Failures**
```python
# Check token format and expiration
token_info = await client.jwt_service.get_token_info(token)
print(f"Token expires: {token_info['expires_at']}")
```
3. **Tenant Not Found**
```python
# Verify tenant GUID and status
tenant = await client.get_tenant_info()
if tenant:
print(f"Tenant status: {tenant.status}")
```
### Debug Logging
Enable debug logging for detailed information:
```python
import logging
logging.getLogger("d3_identity_client").setLevel(logging.DEBUG)
```
## Support
For issues and questions:
- Check the [troubleshooting guide](docs/troubleshooting.md)
- Review [API documentation](docs/api.md)
- Submit issues on GitHub
## License
MIT License - see LICENSE file for details.
Raw data
{
"_id": null,
"home_page": null,
"name": "identityservice-pythonsdk",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "IdentityService Team <support@identityservice.com>",
"keywords": "authentication, jwt, etcd, identity, security, multi-tenant",
"author": "D3 Security",
"author_email": "IdentityService Team <support@identityservice.com>",
"download_url": "https://files.pythonhosted.org/packages/8d/57/7847270e984e341d1df9dcc7ab79cef4259d45a7f1a2ee6b2c0e5ed04cc4/identityservice_pythonsdk-1.0.8.tar.gz",
"platform": null,
"description": "# IdentityService Python SDK\r\n\r\nPython SDK for IdentityService integration, providing multi-tenant authentication, JWT token management, and distributed configuration management.\r\n\r\n## Features\r\n\r\n- **Ed25519 JWT Authentication**: Modern cryptographic signatures for enhanced security\r\n- **Multi-tenant Support**: Tenant-specific configuration and key management\r\n- **Real-time Configuration**: Live configuration updates via etcd watching\r\n- **Service Registration**: Automatic service discovery and health monitoring\r\n- **High-performance Caching**: Memory and Redis-based caching with TTL\r\n- **Framework Integration**: FastAPI and gRPC middleware for seamless integration\r\n- **gRPC TestConnection Service**: Built-in TestConnection service for service health checks\r\n- **Comprehensive Security**: Key rotation, rate limiting, and audit logging\r\n\r\n## Installation\r\n\r\n```bash\r\n# Install from PyPI\r\npip install identityservice-pythonsdk==1.0.6\r\n\r\n# Or install from source\r\npip install -e .\r\n```\r\n\r\n### Requirements\r\n\r\n- Python 3.8+\r\n- etcd cluster (for distributed configuration)\r\n- Redis (optional, for caching)\r\n\r\n## Quick Start\r\n\r\n### 1. Environment Configuration\r\n\r\nSet the following environment variables:\r\n\r\n```bash\r\nexport D3_TENANT_GUID=\"your-tenant-guid\"\r\nexport D3_API_KEY=\"your-tenant-api-key\"\r\nexport ETCD_ENDPOINTS=\"etcd1:2379,etcd2:2379,etcd3:2379\"\r\nexport SERVICE_NAME=\"YourService\"\r\n```\r\n\r\n### 2. Basic Usage\r\n\r\n```python\r\nimport asyncio\r\nfrom d3_identity_client import D3IdentityClient\r\n\r\nasync def main():\r\n # Create and initialize client\r\n async with D3IdentityClient() as client:\r\n # Get tenant information\r\n tenant_info = await client.get_tenant_info()\r\n print(f\"Tenant: {tenant_info.tenant_name}\")\r\n \r\n # Generate service token\r\n token = await client.generate_service_token(\r\n service_name=\"MyService\",\r\n permissions=[\"read\", \"write\"]\r\n )\r\n \r\n # Validate token\r\n result = await client.validate_token(token)\r\n if result.is_valid:\r\n print(f\"Token valid for tenant: {result.claims.tenant_name}\")\r\n\r\nif __name__ == \"__main__\":\r\n asyncio.run(main())\r\n```\r\n\r\n### 3. FastAPI Integration\r\n\r\n```python\r\nfrom fastapi import FastAPI, Depends\r\nfrom d3_identity_client import D3IdentityClient, create_service_auth_dependency\r\n\r\napp = FastAPI()\r\n\r\n# Initialize client (do this at startup)\r\nidentity_client = None\r\n\r\n@app.on_event(\"startup\")\r\nasync def startup():\r\n global identity_client\r\n identity_client = await create_client()\r\n\r\n@app.on_event(\"shutdown\") \r\nasync def shutdown():\r\n if identity_client:\r\n await identity_client.cleanup()\r\n\r\n# Create authentication dependency\r\nauth_required = create_service_auth_dependency(\r\n identity_client.auth_service,\r\n required_service=\"MyService\"\r\n)\r\n\r\n@app.get(\"/protected\")\r\nasync def protected_endpoint(auth_context=auth_required):\r\n return {\r\n \"message\": \"Success\",\r\n \"tenant\": auth_context.tenant_name,\r\n \"service\": auth_context.get_service_name()\r\n }\r\n```\r\n\r\n### 4. gRPC TestConnection Integration\r\n\r\nAdd TestConnection service to your existing CommandService gRPC server:\r\n\r\n```python\r\nimport grpc\r\nfrom concurrent import futures\r\nfrom d3_identity_client import D3IdentityClient, IdentityServiceOptions\r\nfrom d3_identity_client.controllers.grpc_controller import (\r\n TestConnectionGrpcServicer,\r\n add_servicer_to_server\r\n)\r\n\r\nasync def integrate_with_command_service():\r\n # Initialize identity client\r\n config = IdentityServiceOptions(\r\n tenant_guid=\"your-tenant-guid\",\r\n tenant_api_key=\"your-api-key\",\r\n etcd_endpoints=[\"localhost:2379\"],\r\n service_name=\"CommandService\"\r\n )\r\n \r\n identity_client = D3IdentityClient(config)\r\n await identity_client.initialize()\r\n \r\n # Create your existing gRPC server\r\n server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))\r\n \r\n # Add your existing services\r\n # your_pb2_grpc.add_YourServiceServicer_to_server(YourServicer(), server)\r\n \r\n # Add TestConnection service to the SAME server\r\n test_servicer = TestConnectionGrpcServicer(\r\n identity_client.auth_service,\r\n identity_client.tenant_service\r\n )\r\n add_servicer_to_server(test_servicer, server)\r\n \r\n # Start on port 50051 (same port as your existing services)\r\n server.add_insecure_port('[::]:50051')\r\n server.start()\r\n \r\n print(\"\u2705 CommandService with TestConnection running on port 50051\")\r\n return server, identity_client\r\n\r\n# Usage\r\nserver, client = await integrate_with_command_service()\r\n```\r\n\r\n**Call TestConnection from other services:**\r\n\r\n```python\r\nimport grpc\r\nfrom protos import test_connection_pb2, test_connection_pb2_grpc\r\n\r\n# Connect to CommandService (same port for all services)\r\nchannel = grpc.insecure_channel('localhost:50051')\r\nstub = test_connection_pb2_grpc.TestConnectionGrpcServiceStub(channel)\r\n\r\n# Call TestConnection with JWT authentication\r\nmetadata = [('authorization', 'Bearer your-jwt-token')]\r\nrequest = test_connection_pb2.TestConnectionRequest()\r\nresponse = stub.TestConnection(request, metadata=metadata)\r\n\r\nprint(f\"Tenant: {response.TenantName}\")\r\nprint(f\"Status: {response.Status.message}\")\r\n```\r\n\r\n## Configuration\r\n\r\n### Environment Variables\r\n\r\n| Variable | Description | Default |\r\n|----------|-------------|---------|\r\n| `D3_TENANT_GUID` | Tenant GUID | Required |\r\n| `D3_API_KEY` | Tenant API key | Required |\r\n| `ETCD_ENDPOINTS` | etcd endpoints (comma-separated) | `localhost:2379` |\r\n| `ETCD_USERNAME` | etcd authentication username | None |\r\n| `ETCD_PASSWORD` | etcd authentication password | None |\r\n| `CACHE_TYPE` | Cache type: `memory`, `redis`, `hybrid` | `memory` |\r\n| `REDIS_HOST` | Redis host for caching | `localhost` |\r\n| `REDIS_PORT` | Redis port | `6379` |\r\n| `SERVICE_NAME` | Service name for registration | None |\r\n| `LOG_LEVEL` | Logging level | `INFO` |\r\n\r\n### Programmatic Configuration\r\n\r\n```python\r\nfrom d3_identity_client import IdentityServiceOptions, create_production_config\r\n\r\nconfig = create_production_config(\r\n tenant_guid=\"your-tenant-guid\",\r\n api_key=\"your-api-key\",\r\n service_name=\"YourService\",\r\n etcd_endpoints=[\"etcd1:2379\", \"etcd2:2379\"],\r\n redis_host=\"redis.example.com\"\r\n)\r\n\r\nclient = D3IdentityClient(config)\r\n```\r\n\r\n## Advanced Usage\r\n\r\n### Tenant Watching\r\n\r\nMonitor tenant configuration changes in real-time:\r\n\r\n```python\r\nasync def tenant_change_callback(tenant_info):\r\n print(f\"Tenant configuration updated: {tenant_info.tenant_name}\")\r\n # Update application configuration based on tenant changes\r\n\r\nawait client.watch_tenant(tenant_guid, tenant_change_callback)\r\n```\r\n\r\n### Service Discovery\r\n\r\nRegister and discover internal services:\r\n\r\n```python\r\n# Register service\r\nawait client.register_internal_service(\r\n service_name=\"MyService\",\r\n service_endpoints={\r\n \"http\": \"http://myservice:8080\",\r\n \"grpc\": \"grpc://myservice:50051\"\r\n }\r\n)\r\n\r\n# Discover services\r\nservices = await client.get_all_internal_services()\r\ncommand_service = services.get(\"CommandService\")\r\nif command_service:\r\n http_endpoint = command_service.get_endpoint(\"http\")\r\n```\r\n\r\n### Custom Authentication\r\n\r\nImplement custom authentication logic:\r\n\r\n```python\r\nfrom d3_identity_client import AuthService\r\n\r\nasync def custom_auth_handler(token: str):\r\n # Validate token\r\n context = await client.authenticate_token(token)\r\n \r\n if not context:\r\n raise AuthenticationError(\"Invalid token\")\r\n \r\n # Check custom business rules\r\n if not context.token_claims.has_permission(\"custom_action\"):\r\n raise AuthorizationError(\"Insufficient permissions\")\r\n \r\n return context\r\n```\r\n\r\n### Multi-tenant Operations\r\n\r\nWork with multiple tenants:\r\n\r\n```python\r\n# Get multiple tenants\r\ntenant_guids = [\"guid1\", \"guid2\", \"guid3\"]\r\ntenants = await client.get_multiple_tenants(tenant_guids)\r\n\r\nfor guid, tenant in tenants.items():\r\n if tenant and tenant.is_active_tenant():\r\n print(f\"Active tenant: {tenant.tenant_name}\")\r\n```\r\n\r\n## Security Features\r\n\r\n### Key Rotation\r\n\r\nThe client automatically handles key rotation:\r\n\r\n```python\r\n# Keys are automatically rotated by the Identity Service\r\n# Client handles both current and previous keys for validation\r\n# No manual intervention required\r\n```\r\n\r\n### Rate Limiting\r\n\r\nConfigure tenant-specific rate limiting:\r\n\r\n```python\r\ntenant_info = await client.get_tenant_info()\r\napi_limit = tenant_info.get_rate_limit_config(\"api_requests\")\r\n\r\nif api_limit:\r\n print(f\"API rate limit: {api_limit.max_requests_per_minute()}/minute\")\r\n```\r\n\r\n### Permissions\r\n\r\nCheck permissions in your application:\r\n\r\n```python\r\nasync def check_permissions(auth_context, required_permission):\r\n if not await client.auth_service.check_permission(auth_context, required_permission):\r\n raise PermissionError(f\"Permission '{required_permission}' required\")\r\n```\r\n\r\n## Health Monitoring\r\n\r\nMonitor client health:\r\n\r\n```python\r\n@app.get(\"/health\")\r\nasync def health_check():\r\n health_status = await client.health_check()\r\n \r\n if health_status[\"status\"] == \"healthy\":\r\n return {\"status\": \"OK\", \"details\": health_status}\r\n else:\r\n raise HTTPException(status_code=503, detail=\"Service unhealthy\")\r\n```\r\n\r\n## Error Handling\r\n\r\nThe client provides comprehensive error handling:\r\n\r\n```python\r\nfrom d3_identity_client import TokenValidationResult\r\n\r\nasync def safe_token_validation(token: str):\r\n try:\r\n result = await client.validate_token(token)\r\n \r\n if result.is_valid:\r\n return result.claims\r\n else:\r\n logger.warning(f\"Token validation failed: {result.error_message}\")\r\n return None\r\n \r\n except Exception as e:\r\n logger.error(f\"Token validation error: {e}\")\r\n return None\r\n```\r\n\r\n## Development Mode\r\n\r\nFor development and testing:\r\n\r\n```python\r\nfrom d3_identity_client import create_development_config\r\n\r\nconfig = create_development_config(\r\n tenant_guid=\"dev-tenant-guid\",\r\n api_key=\"dev-api-key\",\r\n service_name=\"DevService\"\r\n)\r\n\r\nconfig.debug_mode = True\r\nconfig.logging.log_level = LogLevel.DEBUG\r\n\r\nclient = D3IdentityClient(config)\r\n```\r\n\r\n## Testing\r\n\r\nMock the client for unit testing:\r\n\r\n```python\r\nimport pytest\r\nfrom unittest.mock import AsyncMock\r\n\r\n@pytest.fixture\r\nasync def mock_client():\r\n client = AsyncMock(spec=D3IdentityClient)\r\n client.authenticate_token.return_value = AuthenticationContext(...)\r\n return client\r\n```\r\n\r\n## Migration from Legacy Authentication\r\n\r\nReplace existing JWT authentication:\r\n\r\n```python\r\n# Before (legacy)\r\ndef verify_jwt_token(token):\r\n return jwt.decode(token, secret_key, algorithms=[\"HS256\"])\r\n\r\n# After (D3 Identity Service)\r\nasync def verify_jwt_token(token):\r\n result = await client.validate_token(token)\r\n return result.claims if result.is_valid else None\r\n```\r\n\r\n## Performance Considerations\r\n\r\n- **Caching**: Enable Redis caching for production deployments\r\n- **Connection Pooling**: Client automatically manages etcd connections\r\n- **Token Caching**: JWT validation results are cached for performance\r\n- **Batch Operations**: Use `get_multiple_tenants()` for bulk operations\r\n\r\n## Troubleshooting\r\n\r\n### Common Issues\r\n\r\n1. **etcd Connection Failed**\r\n ```python\r\n # Check etcd endpoints and credentials\r\n health = await client.health_check()\r\n print(health[\"etcd\"])\r\n ```\r\n\r\n2. **Token Validation Failures**\r\n ```python\r\n # Check token format and expiration\r\n token_info = await client.jwt_service.get_token_info(token)\r\n print(f\"Token expires: {token_info['expires_at']}\")\r\n ```\r\n\r\n3. **Tenant Not Found**\r\n ```python\r\n # Verify tenant GUID and status\r\n tenant = await client.get_tenant_info()\r\n if tenant:\r\n print(f\"Tenant status: {tenant.status}\")\r\n ```\r\n\r\n### Debug Logging\r\n\r\nEnable debug logging for detailed information:\r\n\r\n```python\r\nimport logging\r\nlogging.getLogger(\"d3_identity_client\").setLevel(logging.DEBUG)\r\n```\r\n\r\n## Support\r\n\r\nFor issues and questions:\r\n- Check the [troubleshooting guide](docs/troubleshooting.md)\r\n- Review [API documentation](docs/api.md)\r\n- Submit issues on GitHub\r\n\r\n## License\r\n\r\nMIT License - see LICENSE file for details.\r\n",
"bugtrack_url": null,
"license": "MIT License\r\n \r\n Copyright (c) 2025 D3 Security\r\n \r\n Permission is hereby granted, free of charge, to any person obtaining a copy\r\n of this software and associated documentation files (the \"Software\"), to deal\r\n in the Software without restriction, including without limitation the rights\r\n to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\r\n copies of the Software, and to permit persons to whom the Software is\r\n furnished to do so, subject to the following conditions:\r\n \r\n The above copyright notice and this permission notice shall be included in all\r\n copies or substantial portions of the Software.\r\n \r\n THE SOFTWARE IS PROVIDED \"AS IS\", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\r\n IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\r\n FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\r\n AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\r\n LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\r\n OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE\r\n SOFTWARE.",
"summary": "Python SDK for IdentityService integration with multi-tenant authentication, JWT management, and gRPC services",
"version": "1.0.8",
"project_urls": null,
"split_keywords": [
"authentication",
" jwt",
" etcd",
" identity",
" security",
" multi-tenant"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "adb1d99c59b354eb6a71b2fb218ad5b15050c508e9b679c11b4cc7f7494fb619",
"md5": "1444caedd8d2f248cc0954158fe03596",
"sha256": "90e5e5b152dc8c6960daeeb69a8aa2a7ccf70d54806e9b25eafecb4e27c3d5bc"
},
"downloads": -1,
"filename": "identityservice_pythonsdk-1.0.8-py3-none-any.whl",
"has_sig": false,
"md5_digest": "1444caedd8d2f248cc0954158fe03596",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 65184,
"upload_time": "2025-09-13T01:18:00",
"upload_time_iso_8601": "2025-09-13T01:18:00.970326Z",
"url": "https://files.pythonhosted.org/packages/ad/b1/d99c59b354eb6a71b2fb218ad5b15050c508e9b679c11b4cc7f7494fb619/identityservice_pythonsdk-1.0.8-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "8d577847270e984e341d1df9dcc7ab79cef4259d45a7f1a2ee6b2c0e5ed04cc4",
"md5": "8dee7d7b6fa2825ad129f07eb4fd3393",
"sha256": "d0a27d4cfa0adc5c29367f7da33600fa4f042470e79ad21551f72e62ca23630f"
},
"downloads": -1,
"filename": "identityservice_pythonsdk-1.0.8.tar.gz",
"has_sig": false,
"md5_digest": "8dee7d7b6fa2825ad129f07eb4fd3393",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 58080,
"upload_time": "2025-09-13T01:18:02",
"upload_time_iso_8601": "2025-09-13T01:18:02.297135Z",
"url": "https://files.pythonhosted.org/packages/8d/57/7847270e984e341d1df9dcc7ab79cef4259d45a7f1a2ee6b2c0e5ed04cc4/identityservice_pythonsdk-1.0.8.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-09-13 01:18:02",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "identityservice-pythonsdk"
}
