Utility Functions

def generate_temp_url(
    path,
    seconds,
    key,
    method,
    absolute=False,
    prefix_based=False,
    iso8601=False,
    ip_range=None,
    digest=None,
    auth_url=None
):
    """
    Generate a temporary URL for Swift object access.

    Parameters:
    - path: str, object path (e.g., '/v1/AUTH_account/container/object')
    - seconds: int, URL validity duration in seconds from now
    - key: str, secret temporary URL key configured on account/container
    - method: str, HTTP method ('GET', 'PUT', 'POST', 'DELETE')
    - absolute: bool, return absolute timestamp instead of duration (default False)
    - prefix_based: bool, allow access to all objects with path prefix (default False)
    - iso8601: bool, use ISO 8601 timestamp format (default False)
    - ip_range: str, restrict access to IP range (e.g., '192.168.1.0/24')
    - digest: str, digest algorithm ('sha1', 'sha256', 'sha512', default 'sha1')
    - auth_url: str, complete auth URL for absolute URLs

    Returns:
    str: temporary URL with signature and expiration

    Usage:
    Set temp URL key on account: X-Account-Meta-Temp-Url-Key: secret_key
    Or on container: X-Container-Meta-Temp-Url-Key: secret_key
    """

def parse_api_response(headers, body):
    """
    Parse Swift API JSON response.

    Parameters:
    - headers: dict, response headers
    - body: bytes, response body

    Returns:
    list or dict: parsed JSON response, or empty list if no body
    """

def get_body(headers, body):
    """
    Get response body content with proper encoding handling.

    Parameters:
    - headers: dict, response headers
    - body: bytes, raw response body

    Returns:
    str: decoded response body
    """

def config_true_value(value):
    """
    Check if value represents boolean true.

    Parameters:
    - value: any, value to check

    Returns:
    bool: True if value is True or string in ('true', '1', 'yes', 'on', 't', 'y')
    """

def prt_bytes(num_bytes, human_flag):
    """
    Format bytes for human-readable display.

    Parameters:
    - num_bytes: int, number of bytes
    - human_flag: bool, use human-readable format (K, M, G suffixes)

    Returns:
    str: formatted byte string (4 chars for human, 12 chars right-justified otherwise)
    """

def parse_timeout(value):
    """
    Parse timeout strings with suffixes.

    Parameters:
    - value: str, timeout value with optional suffix ('30s', '5m', '2h', '1d')

    Returns:
    float: timeout in seconds

    Supported suffixes:
    - s: seconds
    - m, min: minutes
    - h, hr: hours
    - d: days
    """

def parse_timestamp(seconds, absolute=False):
    """
    Parse timestamp values from various formats.

    Parameters:
    - seconds: str or float, timestamp value or ISO 8601 string
    - absolute: bool, return absolute timestamp (default False for relative)

    Returns:
    float: Unix timestamp

    Raises:
    ValueError: Invalid timestamp format
    """

def split_request_headers(options, prefix=''):
    """
    Split header options into dictionary format.

    Parameters:
    - options: list, header strings in 'name:value' format
    - prefix: str, prefix to add to header names

    Returns:
    dict: headers dictionary with properly formatted names and values
    """

def report_traceback():
    """
    Report exception traceback for debugging.

    Returns:
    str: formatted traceback string
    """

class ReadableToIterable:
    def __init__(self, content, chunk_size=65536, md5=False):
        """
        Convert file-like readable object to iterable.

        Parameters:
        - content: file-like object with read() method
        - chunk_size: int, size of chunks to read (default 65536)
        - md5: bool, calculate MD5 hash while reading (default False)
        """

    def __iter__(self):
        """Iterate over chunks of data."""

    def get_md5sum(self):
        """Get MD5 hash if md5=True was specified."""

class LengthWrapper:
    def __init__(self, readable, length, md5=False):
        """
        Wrap readable object with length limiting.

        Parameters:
        - readable: file-like object with read() method
        - length: int, maximum bytes to read
        - md5: bool, calculate MD5 hash while reading (default False)
        """

    def __iter__(self):
        """Iterate over chunks up to specified length."""

    def read(self, amt=None):
        """Read up to amt bytes or remaining length."""

    def get_md5sum(self):
        """Get MD5 hash if md5=True was specified."""

class NoopMD5:
    """No-operation MD5 hasher for when MD5 is disabled."""

    def update(self, data):
        """No-op update method."""

    def hexdigest(self):
        """Return empty MD5 hash."""
        return ""

class JSONableIterable(list):
    """JSON-serializable iterable that extends list."""

    def __init__(self, iterable=None):
        """Initialize with optional iterable."""

def iter_wrapper(iterable):
    """
    Wrap iterable for streaming uploads.

    Parameters:
    - iterable: any iterable object

    Returns:
    generator: wrapped iterable suitable for streaming
    """

def n_at_a_time(seq, n):
    """
    Split sequence into chunks of size n.

    Parameters:
    - seq: sequence to split
    - n: int, chunk size

    Yields:
    list: chunks of the sequence
    """

def n_groups(seq, n):
    """
    Split sequence into n groups of roughly equal size.

    Parameters:
    - seq: sequence to split
    - n: int, number of groups

    Returns:
    list: list of groups
    """

def normalize_manifest_path(path):
    """
    Normalize manifest paths for consistency.

    Parameters:
    - path: str, manifest path to normalize

    Returns:
    str: normalized path
    """

TRUE_VALUES = {'true', '1', 'yes', 'on', 't', 'y'}
EMPTY_ETAG = 'd41d8cd98f00b204e9800998ecf8427e'
EXPIRES_ISO8601_FORMAT = '%Y-%m-%dT%H:%M:%SZ'
SHORT_EXPIRES_ISO8601_FORMAT = '%Y-%m-%d'
TIME_ERRMSG = 'time must either be a whole number or in specific ISO 8601 format.'

from swiftclient.utils import generate_temp_url
import time

# Set temporary URL key on account (do this once via swift CLI or API)
# swift post -m "Temp-URL-Key:my-secret-key"

# Generate temporary GET URL valid for 1 hour
path = '/v1/AUTH_account/documents/confidential.pdf'
temp_url = generate_temp_url(
    path=path,
    seconds=3600,  # 1 hour
    key='my-secret-key',
    method='GET'
)

print(f"Temporary URL: https://swift.example.com{temp_url}")

# Generate temporary PUT URL for uploads
upload_url = generate_temp_url(
    path='/v1/AUTH_account/uploads/new-file.txt',
    seconds=1800,  # 30 minutes
    key='my-secret-key',
    method='PUT'
)

# Generate prefix-based URL for directory access
prefix_url = generate_temp_url(
    path='/v1/AUTH_account/documents/',
    seconds=7200,  # 2 hours
    key='my-secret-key',
    method='GET',
    prefix_based=True  # Access all objects under documents/
)

# Generate URL with IP restrictions
restricted_url = generate_temp_url(
    path='/v1/AUTH_account/private/data.csv',
    seconds=3600,
    key='my-secret-key',
    method='GET',
    ip_range='192.168.1.0/24'  # Only allow from this subnet
)

# Use absolute timestamp instead of duration
absolute_timestamp = int(time.time()) + 3600  # 1 hour from now
absolute_url = generate_temp_url(
    path=path,
    seconds=absolute_timestamp,
    key='my-secret-key',
    method='GET',
    absolute=True
)

from swiftclient.utils import prt_bytes, config_true_value, parse_timeout

# Format bytes for display
print(prt_bytes(1024, False))      # '        1024'
print(prt_bytes(1024, True))       # '1.0K'
print(prt_bytes(1536000, True))    # '1.5M'
print(prt_bytes(5368709120, True)) # '5.0G'

# Check boolean configuration values
config_values = ['true', '1', 'yes', 'on', 'false', '0', 'no', 'off']
for value in config_values:
    print(f"'{value}' is {config_true_value(value)}")

# Parse timeout strings
timeouts = ['30', '30s', '5m', '2h', '1d']
for timeout in timeouts:
    seconds = parse_timeout(timeout)
    print(f"'{timeout}' = {seconds} seconds")

from swiftclient.utils import split_request_headers

# Parse header options from command line format
header_options = [
    'Content-Type:application/json',
    'X-Object-Meta-Author:John Doe',
    'X-Object-Meta-Version:2.0',
    'Cache-Control:max-age=3600'
]

headers = split_request_headers(header_options)
print(headers)
# {
#     'content-type': 'application/json',
#     'x-object-meta-author': 'John Doe',
#     'x-object-meta-version': '2.0',
#     'cache-control': 'max-age=3600'
# }

# Parse with prefix
container_headers = split_request_headers([
    'Meta-Owner:TeamA',
    'Read:.r:*',
    'Write:.r:*'
], prefix='X-Container-')

print(container_headers)
# {
#     'x-container-meta-owner': 'TeamA',
#     'x-container-read': '.r:*',
#     'x-container-write': '.r:*'
# }

from swiftclient.utils import ReadableToIterable, LengthWrapper
import io

# Convert file to iterable for streaming upload
with open('large_file.dat', 'rb') as f:
    iterable = ReadableToIterable(f, chunk_size=65536, md5=True)

    # Upload using the iterable (example with low-level function)
    etag = put_object(storage_url, token, 'container', 'object', iterable)

    # Get MD5 hash
    md5_hash = iterable.get_md5sum()
    print(f"Uploaded with ETag: {etag}, MD5: {md5_hash}")

# Limit reading to specific length
data = io.BytesIO(b'x' * 10000)  # 10KB of data
limited = LengthWrapper(data, 5000, md5=True)  # Only read first 5KB

chunk_count = 0
for chunk in limited:
    chunk_count += 1
    print(f"Chunk {chunk_count}: {len(chunk)} bytes")

print(f"MD5 of first 5KB: {limited.get_md5sum()}")

from swiftclient.utils import n_at_a_time, n_groups

# Process items in batches
items = list(range(23))

# Split into chunks of 5
for i, chunk in enumerate(n_at_a_time(items, 5)):
    print(f"Batch {i}: {chunk}")
# Batch 0: [0, 1, 2, 3, 4]
# Batch 1: [5, 6, 7, 8, 9]
# ...

# Split into 4 roughly equal groups
groups = n_groups(items, 4)
for i, group in enumerate(groups):
    print(f"Group {i}: {group} ({len(group)} items)")

from swiftclient.utils import parse_api_response, get_body
import json

# Parse Swift API JSON response
headers = {'content-type': 'application/json; charset=utf-8'}
body = json.dumps([
    {'name': 'container1', 'count': 42, 'bytes': 1024000},
    {'name': 'container2', 'count': 17, 'bytes': 512000}
]).encode('utf-8')

containers = parse_api_response(headers, body)
for container in containers:
    print(f"Container: {container['name']}, Objects: {container['count']}")

# Get response body with encoding
text_body = get_body(headers, body)
print(f"Response body: {text_body}")

class ReadableToIterable:
    def __init__(self, content, checksum=None, md5=None, chunk_size=65536):
        """
        Convert file-like objects to iterables suitable for Swift uploads.

        Parameters:
        - content: file-like object or iterable to convert
        - checksum: bool, whether to calculate content checksum
        - md5: hashlib MD5 object, existing MD5 hasher to update
        - chunk_size: int, chunk size for reading (default 65536)

        Yields:
        bytes: content chunks for streaming upload
        """

    def __iter__(self):
        """Return iterator for content chunks."""

    def get_md5sum(self):
        """Get MD5 checksum of processed content."""

class LengthWrapper:
    def __init__(self, readable, length):
        """
        Wrap readable object with known content length.

        Parameters:
        - readable: file-like object or iterable
        - length: int, total content length in bytes

        Used for streaming uploads where content length must be known upfront.
        """

    def __iter__(self):
        """Return iterator for content."""

    def __len__(self):
        """Return content length."""

    def read(self, size=-1):
        """Read up to size bytes from content."""

class JSONableIterable(list):
    def __init__(self, iterable):
        """
        JSON-serializable iterable that preserves iteration behavior.

        Parameters:
        - iterable: any iterable to wrap

        Allows iterables to be JSON serialized while maintaining
        their iteration properties for Swift operations.
        """

class NoopMD5:
    """
    No-operation MD5 hasher for environments where hashlib MD5 is unavailable.

    Provides the same interface as hashlib.md5() but performs no actual hashing.
    Used as fallback when FIPS mode or other restrictions disable MD5.
    """

    def update(self, data):
        """Accept data but perform no hashing."""

    def digest(self):
        """Return empty digest."""

    def hexdigest(self):
        """Return empty hex digest."""

from swiftclient.utils import ReadableToIterable, LengthWrapper

# Convert file to iterable for streaming upload
with open('largefile.dat', 'rb') as f:
    iterable = ReadableToIterable(f, checksum=True)

    # Use in Swift upload
    conn.put_object('container', 'object', iterable)

    # Get checksum after upload
    md5_hash = iterable.get_md5sum()

# Wrap content with known length
content = b'Hello, Swift!'
wrapped = LengthWrapper(iter([content]), len(content))

# Use for precise content length uploads
conn.put_object(
    'container',
    'object',
    wrapped,
    content_length=len(wrapped)
)