tessl/pypi-azure-core

Microsoft Azure Core Library providing foundational infrastructure for Azure SDK Python clients

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

REST API Abstraction

Name: tessl/pypi-azure-core
Author: tessl

Azure Core's REST API abstraction provides a high-level, transport-agnostic interface for making HTTP requests. These classes offer a unified API that works across different HTTP transport implementations while handling content types, serialization, and response processing automatically.

Core Components

HttpRequest

Represents an HTTP request with support for various content types, headers, and parameters.

from azure.core.rest import HttpRequest

class HttpRequest:
    def __init__(
        self,
        method: str,
        url: str,
        *,
        params: Optional[ParamsType] = None,
        headers: Optional[MutableMapping[str, str]] = None,
        json: Any = None,
        content: Optional[ContentType] = None,
        data: Optional[Dict[str, Any]] = None,
        files: Optional[FilesType] = None,
        **kwargs: Any
    ): ...
    
    @property
    def url(self) -> str: ...
    
    @property 
    def method(self) -> str: ...
    
    @property
    def headers(self) -> MutableMapping[str, str]: ...
    
    @property
    def content(self) -> Any: ...

Basic Usage

from azure.core.rest import HttpRequest

# Simple GET request
request = HttpRequest('GET', 'https://api.example.com/data')

# GET with query parameters
request = HttpRequest(
    'GET', 
    'https://api.example.com/search',
    params={'query': 'azure', 'limit': 10}
)

# POST with headers
request = HttpRequest(
    'POST',
    'https://api.example.com/items',
    headers={'Authorization': 'Bearer token'},
    json={'name': 'item1', 'value': 42}
)

Content Types

JSON Content:

request = HttpRequest(
    'POST', 
    '/api/data',
    json={'key': 'value', 'count': 10}
)

Form Data:

request = HttpRequest(
    'POST',
    '/api/form',
    data={'field1': 'value1', 'field2': 'value2'}
)

Raw Content:

# String content
request = HttpRequest('PUT', '/api/text', content='raw text data')

# Binary content
request = HttpRequest('PUT', '/api/binary', content=b'binary data')

# Stream content
request = HttpRequest('PUT', '/api/stream', content=file_stream)

File Uploads:

# Single file
with open('document.pdf', 'rb') as f:
    request = HttpRequest('POST', '/api/upload', files={'file': f})

# Multiple files
files = {
    'doc1': open('file1.txt', 'rb'),
    'doc2': open('file2.txt', 'rb')
}
request = HttpRequest('POST', '/api/upload', files=files)

# File with metadata
request = HttpRequest(
    'POST', 
    '/api/upload',
    files={'file': ('filename.txt', file_content, 'text/plain')}
)

HttpResponse

Abstract base class for HTTP responses with support for various content access patterns and stream management.

from azure.core.rest import HttpResponse

class HttpResponse:
    @property
    def request(self) -> HttpRequest: ...
    
    @property
    def status_code(self) -> int: ...
    
    @property
    def headers(self) -> MutableMapping[str, str]: ...
    
    @property
    def reason(self) -> str: ...
    
    @property
    def content_type(self) -> Optional[str]: ...
    
    @property
    def is_closed(self) -> bool: ...
    
    @property
    def is_stream_consumed(self) -> bool: ...
    
    @property
    def encoding(self) -> Optional[str]: ...
    
    @property
    def url(self) -> str: ...
    
    @property
    def content(self) -> bytes: ...
    
    def text(self, encoding: Optional[str] = None) -> str: ...
    
    def json() -> Any: ...
    
    def raise_for_status() -> None: ...
    
    def close() -> None: ...
    
    def read() -> bytes: ...
    
    def iter_raw(**kwargs) -> Iterator[bytes]: ...
    
    def iter_bytes(**kwargs) -> Iterator[bytes]: ...
    
    def __enter__(self) -> "HttpResponse": ...
    
    def __exit__(self, *args) -> None: ...

Response Usage

Basic Response Handling:

response = client.send_request(request)

# Check status
if response.status_code == 200:
    print("Success!")

# Raise exception for error status codes
response.raise_for_status()

# Access response data
content = response.content  # bytes
text = response.text()      # string
data = response.json()      # parsed JSON

Headers and Metadata:

# Access headers (case-insensitive)
content_type = response.headers['Content-Type']
content_length = response.headers.get('Content-Length')

# Response metadata
print(f"Status: {response.status_code} {response.reason}")
print(f"URL: {response.url}")
print(f"Encoding: {response.encoding}")

Context Manager Usage:

with client.send_request(request) as response:
    response.raise_for_status()
    data = response.json()
    # Response is automatically closed when exiting context

Streaming Responses:

# For large responses, use streaming
request = HttpRequest('GET', '/api/large-file')
response = client.send_request(request, stream=True)

# Iterate over raw bytes (no decompression)
for chunk in response.iter_raw():
    process_raw_chunk(chunk)

# Iterate over bytes (with decompression)
for chunk in response.iter_bytes():
    process_chunk(chunk)

# Always close streaming responses
response.close()

AsyncHttpResponse

Asynchronous version of HttpResponse with async context manager support and async iteration methods.

from azure.core.rest import AsyncHttpResponse

class AsyncHttpResponse(HttpResponse):
    async def read() -> bytes: ...
    
    async def iter_raw(**kwargs) -> AsyncIterator[bytes]: ...
    
    async def iter_bytes(**kwargs) -> AsyncIterator[bytes]: ...
    
    async def close() -> None: ...
    
    async def __aenter__(self) -> "AsyncHttpResponse": ...
    
    async def __aexit__(self, *args) -> None: ...

Async Usage

Basic Async Response Handling:

async def handle_response():
    response = await async_client.send_request(request)
    
    # Async context manager
    async with async_client.send_request(request) as response:
        response.raise_for_status()
        content = await response.read()
        return response.json()

Async Streaming:

async def stream_large_file():
    request = HttpRequest('GET', '/api/large-download')
    response = await async_client.send_request(request, stream=True)
    
    try:
        async for chunk in response.iter_bytes():
            await process_chunk(chunk)
    finally:
        await response.close()

Type Definitions

from typing import Union, Optional, Mapping, Sequence, Iterable, AsyncIterable, IO, Tuple

# Content types
ContentType = Union[str, bytes, Iterable[bytes], AsyncIterable[bytes]]

# Parameter types for query strings
PrimitiveData = Optional[Union[str, int, float, bool]]
ParamsType = Mapping[str, Union[PrimitiveData, Sequence[PrimitiveData]]]

# File upload types
FileContent = Union[str, bytes, IO[str], IO[bytes]]
FileType = Union[
    FileContent,
    Tuple[Optional[str], FileContent],  # (filename, content)
    Tuple[Optional[str], FileContent, Optional[str]]  # (filename, content, content_type)
]
FilesType = Union[Mapping[str, FileType], Sequence[Tuple[str, FileType]]]

Integration with PipelineClient

The REST API classes are designed to work seamlessly with PipelineClient's send_request method:

from azure.core import PipelineClient
from azure.core.rest import HttpRequest

client = PipelineClient(base_url="https://api.example.com")

# Create request using REST API abstraction
request = HttpRequest('GET', '/users', params={'page': 1})

# Send through pipeline
response = client.send_request(request)

# Use REST response methods
response.raise_for_status()
users = response.json()

Key Advantages

Transport Agnostic: Works with any HTTP transport (requests, aiohttp, etc.) without changing your code.

Automatic Content Handling: Automatically sets appropriate headers and serializes content based on the type you provide.

Unified Interface: Consistent API across synchronous and asynchronous operations.

Stream Management: Proper handling of response streams with automatic resource cleanup.

Error Handling: Built-in status code checking with raise_for_status().

Backward Compatibility: Maintains compatibility with older Azure SDK patterns while providing modern conveniences.

The REST API abstraction simplifies HTTP operations while maintaining the full power and flexibility of the underlying pipeline system, making it the recommended approach for most HTTP interactions in Azure SDK applications.

Install with Tessl CLI