tessl/pypi-msrest
AutoRest swagger generator Python client runtime for REST API clients with serialization, authentication, and request handling.
—
Pending
service-client.mddocs/
Service Client
Core REST client functionality providing HTTP request creation, pipeline processing, URL formatting, streaming support, and context management. The ServiceClient class is the primary interface for sending HTTP requests through the msrest pipeline system.
Capabilities
Client Creation
Create service clients with configuration and optional credentials for REST API communication.
class ServiceClient:
def __init__(self, creds, config: Configuration):
"""
Initialize REST service client.
Parameters:
- creds: Authentication credentials (deprecated, use config.credentials)
- config: Service configuration object
"""
class SDKClient:
def __init__(self, creds, config: Configuration):
"""
Base class for all generated SDK clients.
Parameters:
- creds: Authentication credentials
- config: Service configuration object
"""
def close(self):
"""Close the client if keep_alive is True."""HTTP Request Methods
Create HTTP requests for different HTTP verbs with parameters, headers, and body content.
def get(self, url: str, params=None, headers=None, content=None, form_content=None):
"""
Create a GET request object.
Parameters:
- url: The request URL
- params: URL query parameters dict
- headers: HTTP headers dict
- content: Request body content
- form_content: Form data dict
Returns:
ClientRequest object
"""
def post(self, url: str, params=None, headers=None, content=None, form_content=None):
"""
Create a POST request object.
Parameters:
- url: The request URL
- params: URL query parameters dict
- headers: HTTP headers dict
- content: Request body content
- form_content: Form data dict
Returns:
ClientRequest object
"""
def put(self, url: str, params=None, headers=None, content=None, form_content=None):
"""Create a PUT request object."""
def patch(self, url: str, params=None, headers=None, content=None, form_content=None):
"""Create a PATCH request object."""
def delete(self, url: str, params=None, headers=None, content=None, form_content=None):
"""Create a DELETE request object."""
def head(self, url: str, params=None, headers=None, content=None, form_content=None):
"""Create a HEAD request object."""
def merge(self, url: str, params=None, headers=None, content=None, form_content=None):
"""Create a MERGE request object."""Request Sending
Send prepared requests through the configured pipeline with response handling.
def send(self, request, headers=None, content=None, **kwargs):
"""
Send request object according to configuration.
Parameters:
- request: ClientRequest object to send
- headers: Additional headers dict (deprecated)
- content: Additional body content (deprecated)
- kwargs: Additional configuration overrides
Returns:
HTTP response object
"""
def send_formdata(self, request, headers=None, content=None, **config):
"""
Send data as multipart form-data request (deprecated).
Parameters:
- request: ClientRequest object
- headers: Additional headers
- content: Form data dict
- config: Configuration overrides
"""URL Formatting
Format request URLs with the client base URL and parameter substitution.
def format_url(self, url: str, **kwargs) -> str:
"""
Format request URL with client base URL.
Parameters:
- url: Request URL to format
- kwargs: URL template parameters
Returns:
Formatted absolute URL string
"""Streaming Support
Handle streaming uploads and downloads for large data transfers.
def stream_upload(self, data, callback):
"""
Generator for streaming request body data.
Parameters:
- data: File-like object to stream
- callback: Progress monitoring callback
Yields:
Data chunks for upload
"""
def stream_download(self, data, callback):
"""
Generator for streaming response body data.
Parameters:
- data: Response object to stream
- callback: Progress monitoring callback
Returns:
Iterator of response data chunks
"""Context Management
Support for using service clients as context managers for automatic resource cleanup.
def __enter__(self):
"""Enter context manager, enable keep_alive."""
def __exit__(self, *exc_details):
"""Exit context manager, disable keep_alive."""
def close(self):
"""Close the pipeline if keep_alive is True."""Header Management
Add persistent headers that apply to all requests (deprecated method).
def add_header(self, header: str, value: str):
"""
Add persistent header (deprecated).
Use config.headers instead.
Parameters:
- header: Header name
- value: Header value
"""Usage Examples
Basic Client Usage
from msrest import ServiceClient, Configuration
# Create configuration
config = Configuration(base_url='https://api.example.com')
# Create and use client
with ServiceClient(None, config) as client:
# Create GET request
request = client.get('/users', params={'limit': 10})
# Send request
response = client.send(request)
# Process response
print(f"Status: {response.status_code}")
print(f"Data: {response.text}")POST Request with JSON Body
import json
from msrest import ServiceClient, Configuration
config = Configuration(base_url='https://api.example.com')
with ServiceClient(None, config) as client:
# Prepare JSON data
data = {'name': 'John Doe', 'email': 'john@example.com'}
json_data = json.dumps(data)
# Create POST request
request = client.post('/users',
content=json_data,
headers={'Content-Type': 'application/json'})
# Send request
response = client.send(request)
print(f"Created user: {response.status_code}")Streaming Upload
from msrest import ServiceClient, Configuration
config = Configuration(base_url='https://api.example.com')
with ServiceClient(None, config) as client:
# Define progress callback
def progress_callback(chunk, response=None):
print(f"Uploaded {len(chunk)} bytes")
# Stream file upload
with open('large_file.dat', 'rb') as f:
for chunk in client.stream_upload(f, progress_callback):
# Chunks are yielded for upload
passTypes
class ClientRequest:
"""Universal HTTP request abstraction."""
method: str
url: str
headers: dict
data: any
files: dictInstall with Tessl CLI
npx tessl i tessl/pypi-msrest