Python SDK

Installation

pip

pip install qodryx

Poetry

poetry add qodryx

With async support

pip install qodryx[async]

Requirements

• Python 3.8 or higher
• httpx (for HTTP requests)
• pydantic (for data validation)

Quick Start

from qodryx import Qodryx
import os

# Initialize the client
client = Qodryx(api_key=os.environ['QODRYX_API_KEY'])

# List projects
projects = client.projects.list()
for project in projects.data:
    print(f"{project.name}: {project.id}")

# Trigger a security scan
scan = client.security.scan(
    project_id='proj_abc123',
    types=['sast', 'secrets', 'dependencies']
)
print(f"Scan started: {scan.id}")

Async Usage

import asyncio
from qodryx import AsyncQodryx
import os

async def main():
    async with AsyncQodryx(api_key=os.environ['QODRYX_API_KEY']) as client:
        # All methods are async
        projects = await client.projects.list()
        
        # Run multiple operations concurrently
        scans = await asyncio.gather(*[
            client.security.scan(project_id=p.id)
            for p in projects.data
        ])
        
        for scan in scans:
            print(f"Started scan: {scan.id}")

asyncio.run(main())

Configuration

from qodryx import Qodryx

client = Qodryx(
    # Required: Your API key
    api_key=os.environ['QODRYX_API_KEY'],
    
    # Optional: Base URL (default: https://api.qodryx.com)
    base_url='https://api.qodryx.com',
    
    # Optional: Request timeout in seconds (default: 30)
    timeout=60,
    
    # Optional: Retry configuration
    max_retries=3,
    retry_on_rate_limit=True,
    
    # Optional: Custom headers
    headers={'X-Custom-Header': 'value'},
)

Type Hints

The SDK provides full type hints for better IDE support:

from qodryx import Qodryx
from qodryx.types import Project, SecurityScan, Workflow

client = Qodryx(api_key=os.environ['QODRYX_API_KEY'])

# IDE autocomplete and type checking work out of the box
project: Project = client.projects.get('proj_abc123')
scan: SecurityScan = client.security.scan(project_id=project.id)
workflow: Workflow = client.workflows.create(
    project_id=project.id,
    prompt='Add user authentication'
)

API Reference

Projects

# List all projects
projects = client.projects.list(page=1, per_page=20)

# Get a specific project
project = client.projects.get('proj_abc123')

# Create a new project
new_project = client.projects.create(
    name='my-app',
    repository='github.com/user/my-app',
    description='My awesome app'
)

# Update a project
updated = client.projects.update('proj_abc123', name='new-name')

# Delete a project
client.projects.delete('proj_abc123')

Workflows

# Create and trigger a workflow
workflow = client.workflows.create(
    project_id='proj_abc123',
    prompt='Add OAuth authentication with Google',
    config={
        'ai_provider': 'anthropic',
        'auto_deploy': False
    }
)

# List workflows
workflows = client.workflows.list(
    project_id='proj_abc123',
    status='running'
)

# Get workflow status
status = client.workflows.get('wf_xyz789')

# Get workflow logs
logs = client.workflows.logs('wf_xyz789', stage='build')

# Cancel a workflow
client.workflows.cancel('wf_xyz789')

# Wait for workflow completion
result = client.workflows.wait_for_completion(
    'wf_xyz789',
    timeout=300,  # 5 minutes
    poll_interval=5  # 5 seconds
)

Security

# Trigger a security scan
scan = client.security.scan(
    project_id='proj_abc123',
    types=['sast', 'secrets', 'dependencies'],
    branch='main'
)

# Get scan results
results = client.security.results('scan_abc123')

# List all findings
findings = client.security.findings(
    project_id='proj_abc123',
    severity=['critical', 'high'],
    status='open'
)

# Get specific finding
finding = client.security.finding('finding_abc123')

# Auto-remediate a finding
remediation = client.security.remediate('finding_abc123')

# Ignore a finding
client.security.ignore(
    'finding_abc123',
    reason='False positive - test file'
)

Deployments

# Trigger a deployment
deployment = client.deployments.create(
    project_id='proj_abc123',
    environment='production',
    provider='vercel',
    branch='main'
)

# Get deployment status
status = client.deployments.get('deploy_abc123')

# List deployments
deployments = client.deployments.list(
    project_id='proj_abc123',
    environment='production'
)

# Rollback a deployment
client.deployments.rollback('deploy_abc123')

# Get deployment logs
logs = client.deployments.logs('deploy_abc123')

Tests

# Generate tests for a file
tests = client.tests.generate(
    project_id='proj_abc123',
    file_path='src/auth/login.py',
    framework='pytest'
)

# Run tests
results = client.tests.run(
    project_id='proj_abc123',
    coverage=True
)

# Get test coverage
coverage = client.tests.coverage('proj_abc123')

Error Handling

from qodryx import Qodryx
from qodryx.exceptions import (
    QodrxyError,
    AuthenticationError,
    RateLimitError,
    NotFoundError,
    ValidationError
)

client = Qodryx(api_key=os.environ['QODRYX_API_KEY'])

try:
    scan = client.security.scan(project_id='proj_abc123')
except AuthenticationError:
    print('Invalid API key')
except RateLimitError as e:
    print(f'Rate limit exceeded, retry after {e.retry_after} seconds')
except NotFoundError:
    print('Project not found')
except ValidationError as e:
    print(f'Validation error: {e.errors}')
except QodrxyError as e:
    print(f'API error: {e.code} - {e.message}')

Context Manager

Use the context manager for automatic resource cleanup:

from qodryx import Qodryx

with Qodryx(api_key=os.environ['QODRYX_API_KEY']) as client:
    projects = client.projects.list()
    # Connection is automatically closed after the block

# Async context manager
async with AsyncQodryx(api_key=os.environ['QODRYX_API_KEY']) as client:
    projects = await client.projects.list()

Pagination

# Manual pagination
page = 1
has_more = True

while has_more:
    response = client.projects.list(page=page, per_page=50)
    for project in response.data:
        print(project.name)
    
    has_more = page < response.meta.total_pages
    page += 1

# Auto-pagination iterator
for project in client.projects.list_all():
    print(project.name)

# Async auto-pagination
async for project in client.projects.list_all():
    print(project.name)

Webhooks

Verify and parse webhook payloads:

from qodryx.webhooks import verify_webhook, parse_event
from flask import Flask, request

app = Flask(__name__)

@app.route('/webhooks/qodryx', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Qodryx-Signature')
    payload = request.data
    
    # Verify signature
    if not verify_webhook(payload, signature, WEBHOOK_SECRET):
        return 'Invalid signature', 401
    
    # Parse event
    event = parse_event(request.json)
    
    if event.type == 'workflow.completed':
        print(f'Workflow completed: {event.data.workflow_id}')
    elif event.type == 'security.finding.created':
        print(f'New vulnerability: {event.data.severity}')
    
    return 'OK', 200

Django Integration

# settings.py
QODRYX_API_KEY = os.environ['QODRYX_API_KEY']

# views.py
from django.conf import settings
from qodryx import Qodryx

def security_check(request, project_id):
    client = Qodryx(api_key=settings.QODRYX_API_KEY)
    
    # Trigger scan
    scan = client.security.scan(
        project_id=project_id,
        types=['sast', 'secrets']
    )
    
    return JsonResponse({'scan_id': scan.id})

FastAPI Integration

from fastapi import FastAPI, Depends
from qodryx import AsyncQodryx
import os

app = FastAPI()

async def get_qodryx_client():
    async with AsyncQodryx(api_key=os.environ['QODRYX_API_KEY']) as client:
        yield client

@app.get('/projects')
async def list_projects(client: AsyncQodryx = Depends(get_qodryx_client)):
    projects = await client.projects.list()
    return {'projects': [p.dict() for p in projects.data]}

@app.post('/scan/{project_id}')
async def trigger_scan(
    project_id: str,
    client: AsyncQodryx = Depends(get_qodryx_client)
):
    scan = await client.security.scan(project_id=project_id)
    return {'scan_id': scan.id}

Examples

Complete Example: Automated Security Check

#!/usr/bin/env python3
"""
Automated security check script for CI/CD pipelines
"""
import os
import sys
import time
from qodryx import Qodryx
from qodryx.exceptions import QodrxyError

def run_security_check(project_id: str) -> bool:
    """Run security scan and return True if no critical issues."""
    client = Qodryx(api_key=os.environ['QODRYX_API_KEY'])
    
    print('Starting security scan...')
    
    # Trigger scan
    scan = client.security.scan(
        project_id=project_id,
        types=['sast', 'secrets', 'dependencies']
    )
    print(f'Scan started: {scan.id}')
    
    # Wait for completion
    while True:
        results = client.security.results(scan.id)
        
        if results.status == 'completed':
            print('Scan completed!')
            
            # Check for critical findings
            critical = [f for f in results.findings if f.severity == 'critical']
            
            if critical:
                print(f'❌ Found {len(critical)} critical vulnerabilities:')
                for finding in critical:
                    print(f'  - {finding.title} ({finding.file}:{finding.line})')
                return False
            
            print('✅ No critical vulnerabilities found')
            return True
        
        if results.status == 'failed':
            print('❌ Scan failed')
            return False
        
        print(f'Scan in progress... ({results.status})')
        time.sleep(5)


if __name__ == '__main__':
    project_id = sys.argv[1] if len(sys.argv) > 1 else os.environ.get('QODRYX_PROJECT')
    
    if not project_id:
        print('Usage: python security_check.py <project_id>')
        sys.exit(1)
    
    try:
        success = run_security_check(project_id)
        sys.exit(0 if success else 1)
    except QodrxyError as e:
        print(f'Error: {e.message}')
        sys.exit(1)

Next Steps