mypy-boto3-mediastore-data

import boto3 from mypy_boto3_mediastore_data.client import MediaStoreDataClient from mypy_boto3_mediastore_data.type_defs import PutObjectResponseTypeDef def upload_to_mediastore_data(container_endpoint: str, object_path: str, data: bytes, content_type: str) -> PutObjectResponseTypeDef: """Uploads an object to AWS Elemental MediaStore Data plane.""" # Create a boto3 session. Credentials will be picked up from environment variables or ~/.aws/credentials session = boto3.Session() # Create a MediaStoreData client using the container endpoint client: MediaStoreDataClient = session.client( "mediastore-data", endpoint_url=container_endpoint, region_name=os.environ.get('AWS_REGION', 'us-east-1') # Or specify your region ) print(f"Uploading object to {object_path} in container at {container_endpoint}...") response: PutObjectResponseTypeDef = client.put_object( Body=data, Path=object_path, ContentType=content_type, # StorageClass='TEMPORAL' # Optional, default is TEMPORAL # UploadAvailability='STANDARD' # Optional, default is STANDARD ) print(f"Upload successful. ETag: {response.get('ETag')}, ContentSHA256: {response.get('ContentSHA256')}") return response # Example usage (replace with your actual endpoint and data) if __name__ == "__main__": import os sample_endpoint = os.environ.get('MEDIASTORE_DATA_ENDPOINT', 'https://data.mediastore.us-east-1.amazonaws.com') sample_path = "videos/my_video.mp4" sample_data = b"This is some sample video content." sample_content_type = "video/mp4" if not sample_endpoint or sample_endpoint == 'https://data.mediastore.us-east-1.amazonaws.com': print("WARNING: Please set the MEDIASTORE_DATA_ENDPOINT environment variable to your container's data endpoint.") print("Skipping actual upload, showing example structure only.") else: try: upload_to_mediastore_data(sample_endpoint, sample_path, sample_data, sample_content_type) except Exception as e: print(f"Error during upload: {e}")