tessl/pypi-python-swiftclient
OpenStack Object Storage API Client Library
low-level-operations.mddocs/
Low-Level Operations
Direct HTTP-level functions for Swift operations providing fine-grained control over requests, custom headers, and response handling. These functions operate at the HTTP level and require manual authentication and connection management.
Capabilities
Account Operations
Direct HTTP operations for account-level functionality including metadata retrieval and container listings.
def get_account(
url,
token,
marker=None,
limit=None,
prefix=None,
end_marker=None,
http_conn=None,
full_listing=False,
service_token=None,
headers=None,
delimiter=None
):
"""
Get a listing of containers for the account.
Parameters:
- url: str, storage URL
- token: str, auth token
- marker: str, marker query for pagination
- limit: int, limit query for number of containers
- prefix: str, prefix query to filter container names
- end_marker: str, end marker query for pagination
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- full_listing: bool, return complete listing with multiple requests if needed
- service_token: str, service auth token
- headers: dict, additional headers to include
- delimiter: str, delimiter query for hierarchical listings
Returns:
tuple: (response_headers dict, containers list)
Raises:
ClientException: HTTP GET request failed
"""
def head_account(url, token, http_conn=None, headers=None, service_token=None):
"""
Get account stats and metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- headers: dict, additional headers to include
- service_token: str, service auth token
Returns:
dict: response headers with account metadata (all lowercase keys)
Raises:
ClientException: HTTP HEAD request failed
"""
def post_account(
url,
token,
headers,
http_conn=None,
response_dict=None,
service_token=None,
query_string=None,
data=None
):
"""
Update account metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- headers: dict, headers to set as account metadata
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
- query_string: str, query string to append to path
- data: bytes, optional message body for request
Returns:
tuple: (response_headers dict, response_body bytes)
Raises:
ClientException: HTTP POST request failed
"""Container Operations
Direct HTTP operations for container management including creation, listing, metadata, and deletion.
def get_container(
url,
token,
container,
marker=None,
limit=None,
prefix=None,
delimiter=None,
end_marker=None,
version_marker=None,
path=None,
http_conn=None,
full_listing=False,
service_token=None,
headers=None,
query_string=None
):
"""
Get a listing of objects for the container.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name to get listing for
- marker: str, marker query for pagination
- limit: int, limit query for number of objects
- prefix: str, prefix query to filter object names
- delimiter: str, delimiter to group objects hierarchically
- end_marker: str, end marker query for pagination
- version_marker: str, version marker query for versioned objects
- path: str, path query (equivalent to delimiter='/' and prefix=path/)
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- full_listing: bool, return complete listing with multiple requests if needed
- service_token: str, service auth token
- headers: dict, additional headers to include
- query_string: str, query string to append to path
Returns:
tuple: (response_headers dict, objects list)
Raises:
ClientException: HTTP GET request failed
"""
def head_container(url, token, container, http_conn=None, headers=None, service_token=None):
"""
Get container stats and metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name to get stats for
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- headers: dict, additional headers to include
- service_token: str, service auth token
Returns:
dict: response headers with container metadata (all lowercase keys)
Raises:
ClientException: HTTP HEAD request failed
"""
def put_container(
url,
token,
container,
headers=None,
http_conn=None,
response_dict=None,
service_token=None,
query_string=None
):
"""
Create a container.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name to create
- headers: dict, additional headers to set on container
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
- query_string: str, query string to append to path
Raises:
ClientException: HTTP PUT request failed
"""
def post_container(
url,
token,
container,
headers,
http_conn=None,
response_dict=None,
service_token=None
):
"""
Update container metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name to update
- headers: dict, headers to set as container metadata
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
Raises:
ClientException: HTTP POST request failed
"""
def delete_container(
url,
token,
container,
http_conn=None,
response_dict=None,
service_token=None,
query_string=None,
headers=None
):
"""
Delete a container.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name to delete
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
- query_string: str, query string to append to path
- headers: dict, additional headers to include
Raises:
ClientException: HTTP DELETE request failed
"""Object Operations
Direct HTTP operations for object management including upload, download, metadata, copying, and deletion.
def get_object(
url,
token,
container,
name,
http_conn=None,
resp_chunk_size=None,
query_string=None,
response_dict=None,
headers=None,
service_token=None
):
"""
Get an object.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name that the object is in
- name: str, object name to get
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- resp_chunk_size: int, chunk size for reading response (enables streaming)
- query_string: str, query string to append to path
- response_dict: dict, optional dict to store response info
- headers: dict, additional headers to include (e.g., Range)
- service_token: str, service auth token
Returns:
tuple: (response_headers dict, object_content bytes or iterable)
Raises:
ClientException: HTTP GET request failed
"""
def head_object(
url,
token,
container,
name,
http_conn=None,
service_token=None,
headers=None,
query_string=None
):
"""
Get object info and metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name that the object is in
- name: str, object name to get info for
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- service_token: str, service auth token
- headers: dict, additional headers to include
- query_string: str, query string to append to path
Returns:
dict: response headers with object metadata (all lowercase keys)
Raises:
ClientException: HTTP HEAD request failed
"""
def put_object(
url,
token=None,
container=None,
name=None,
contents=None,
content_length=None,
etag=None,
chunk_size=None,
content_type=None,
headers=None,
http_conn=None,
proxy=None,
query_string=None,
response_dict=None,
service_token=None
):
"""
Put an object.
Parameters:
- url: str, storage URL
- token: str, auth token (None to send no token)
- container: str, container name (None if part of URL)
- name: str, object name (None if part of URL)
- contents: str/bytes/file-like/iterable, object data (None for zero-byte PUT)
- content_length: int, content length (computed if None)
- etag: str, MD5 hash of contents (None to not send)
- chunk_size: int, chunk size for file-like objects (default 65536)
- content_type: str, content type header value
- headers: dict, additional headers to include
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- proxy: str, proxy URL (format: 'http://127.0.0.1:8888')
- query_string: str, query string to append to path
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
Returns:
str: ETag of uploaded object
Raises:
ClientException: HTTP PUT request failed
"""
def post_object(
url,
token,
container,
name,
headers,
http_conn=None,
response_dict=None,
service_token=None
):
"""
Update object metadata.
Parameters:
- url: str, storage URL
- token: str, auth token
- container: str, container name that the object is in
- name: str, name of the object to update
- headers: dict, headers to set as object metadata
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
Raises:
ClientException: HTTP POST request failed
"""
def copy_object(
url,
token,
container,
name,
destination=None,
headers=None,
fresh_metadata=None,
http_conn=None,
response_dict=None,
service_token=None
):
"""
Copy object.
Parameters:
- url: str, storage URL
- token: str, auth token (None to send no token)
- container: str, container name that the source object is in
- name: str, source object name
- destination: str, destination in format /container/object (None uses source)
- headers: dict, additional headers to include
- fresh_metadata: bool, omit existing user metadata (None/False preserves)
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
Raises:
ClientException: HTTP COPY request failed
"""
def delete_object(
url,
token=None,
container=None,
name=None,
http_conn=None,
headers=None,
proxy=None,
query_string=None,
response_dict=None,
service_token=None
):
"""
Delete object.
Parameters:
- url: str, storage URL
- token: str, auth token (None to send no token)
- container: str, container name (None if part of URL)
- name: str, object name (None if part of URL)
- http_conn: tuple, (parsed_url, HTTPConnection) or None for new connection
- headers: dict, additional headers to include
- proxy: str, proxy URL (format: 'http://127.0.0.1:8888')
- query_string: str, query string to append to path
- response_dict: dict, optional dict to store response info
- service_token: str, service auth token
Raises:
ClientException: HTTP DELETE request failed
"""Connection and Capability Functions
Low-level connection management and cluster information retrieval.
def http_connection(*args, **kwargs):
"""
Create HTTP connection for Swift requests.
Returns:
tuple: (parsed_url, HTTPConnection_object)
"""
def get_capabilities(http_conn):
"""
Get cluster capability information.
Parameters:
- http_conn: tuple, (parsed_url, HTTPConnection)
Returns:
dict: cluster capabilities and configuration
Raises:
ClientException: HTTP Capabilities GET failed
"""Usage Examples
Basic Low-Level Usage
from swiftclient import (
get_auth, get_account, put_container, put_object, get_object, delete_object
)
# Authenticate first
storage_url, token = get_auth(
'https://identity.example.com:5000/v3',
'username',
'password',
auth_version='3',
os_options={'project_name': 'project-name'}
)
# Account operations
account_headers = head_account(storage_url, token)
print(f"Account containers: {account_headers['x-account-container-count']}")
headers, containers = get_account(storage_url, token, limit=10)
for container in containers:
print(f"Container: {container['name']}")
# Container operations
put_container(storage_url, token, 'documents')
headers, objects = get_container(storage_url, token, 'documents')
# Object operations
etag = put_object(
storage_url,
token,
'documents',
'hello.txt',
'Hello, World!',
content_type='text/plain'
)
headers, content = get_object(storage_url, token, 'documents', 'hello.txt')
print(content.decode('utf-8'))
delete_object(storage_url, token, 'documents', 'hello.txt')Connection Reuse
from swiftclient import http_connection, get_account, get_container
# Create persistent connection
storage_url, token = get_auth(...)
parsed_url, conn = http_connection(storage_url)
try:
# Reuse connection for multiple operations
headers, containers = get_account(
storage_url, token, http_conn=(parsed_url, conn)
)
for container in containers:
headers, objects = get_container(
storage_url, token, container['name'], http_conn=(parsed_url, conn)
)
print(f"Container {container['name']} has {len(objects)} objects")
finally:
conn.close()Streaming Large Objects
import io
# Upload large object with streaming
large_data = io.BytesIO(b'x' * 50000000) # 50MB
etag = put_object(
storage_url,
token,
'documents',
'large-file.dat',
large_data,
content_length=50000000,
chunk_size=1048576 # 1MB chunks
)
# Download with streaming
headers, content = get_object(
storage_url,
token,
'documents',
'large-file.dat',
resp_chunk_size=1048576 # 1MB chunks
)
total_size = 0
for chunk in content:
total_size += len(chunk)
print(f"Downloaded {total_size} bytes", end='\r')
print(f"\nTotal: {total_size} bytes")Custom Headers and Metadata
# Upload with custom headers
custom_headers = {
'X-Object-Meta-Author': 'John Doe',
'X-Object-Meta-Project': 'Documentation',
'Content-Disposition': 'attachment; filename="report.pdf"',
'Cache-Control': 'max-age=3600'
}
put_object(
storage_url,
token,
'documents',
'report.pdf',
pdf_content,
content_type='application/pdf',
headers=custom_headers
)
# Retrieve metadata
headers = head_object(storage_url, token, 'documents', 'report.pdf')
print(f"Author: {headers.get('x-object-meta-author')}")
print(f"Last modified: {headers.get('last-modified')}")
print(f"Content length: {headers.get('content-length')}")
# Update metadata only
post_object(
storage_url,
token,
'documents',
'report.pdf',
{'X-Object-Meta-Version': '2.0'}
)Range Requests
# Download specific byte range
range_headers = {'Range': 'bytes=0-1023'} # First 1KB
headers, partial_content = get_object(
storage_url,
token,
'documents',
'large-file.dat',
headers=range_headers
)
print(f"Content range: {headers.get('content-range')}")
print(f"Partial content length: {len(partial_content)}")Error Handling
from swiftclient import ClientException
try:
headers, content = get_object(storage_url, token, 'documents', 'nonexistent.txt')
except ClientException as e:
if e.http_status == 404:
print("Object not found")
elif e.http_status == 401:
print("Authentication failed")
else:
print(f"Request failed: {e.http_status} {e.http_reason}")Install with Tessl CLI
npx tessl i tessl/pypi-python-swiftclient