def presigned_put_object(
self,
bucket: "capo_s3.types.bucket_name.BucketName",
key: "capo_s3.types.object_key.ObjectKey",
expire_in: int = 3600,
*,
config_overrides: Optional[S3ClientConfig] = None,
acl: Optional["capo_s3.types.object_canned_acl.ObjectCannedACL"] = None,
cache_control: Optional["capo_s3.types.cache_control.CacheControl"] = None,
content_disposition: Optional[
"capo_s3.types.content_disposition.ContentDisposition"
] = None,
content_encoding: Optional[
"capo_s3.types.content_encoding.ContentEncoding"
] = None,
content_language: Optional[
"capo_s3.types.content_language.ContentLanguage"
] = None,
content_length: Optional["capo_s3.types.content_length.ContentLength"] = None,
content_md5: Optional["capo_s3.types.content_md5.ContentMD5"] = None,
content_type: Optional["capo_s3.types.content_type.ContentType"] = None,
checksum_algorithm: Optional[
"capo_s3.types.checksum_algorithm.ChecksumAlgorithm"
] = None,
checksum_crc32: Optional["capo_s3.types.checksum_crc32.ChecksumCRC32"] = None,
checksum_crc32_c: Optional[
"capo_s3.types.checksum_crc32_c.ChecksumCRC32C"
] = None,
checksum_crc64_nvme: Optional[
"capo_s3.types.checksum_crc64_nvme.ChecksumCRC64NVME"
] = None,
checksum_sha1: Optional["capo_s3.types.checksum_sha1.ChecksumSHA1"] = None,
checksum_sha256: Optional[
"capo_s3.types.checksum_sha256.ChecksumSHA256"
] = None,
checksum_sha512: Optional[
"capo_s3.types.checksum_sha512.ChecksumSHA512"
] = None,
checksum_md5: Optional["capo_s3.types.checksum_md5.ChecksumMD5"] = None,
checksum_xxhash64: Optional[
"capo_s3.types.checksum_xxhash64.ChecksumXXHASH64"
] = None,
checksum_xxhash3: Optional[
"capo_s3.types.checksum_xxhash3.ChecksumXXHASH3"
] = None,
checksum_xxhash128: Optional[
"capo_s3.types.checksum_xxhash128.ChecksumXXHASH128"
] = None,
expires: Optional["capo_s3.types.expires.Expires"] = None,
if_match: Optional["capo_s3.types.if_match.IfMatch"] = None,
if_none_match: Optional["capo_s3.types.if_none_match.IfNoneMatch"] = None,
grant_full_control: Optional[
"capo_s3.types.grant_full_control.GrantFullControl"
] = None,
grant_read: Optional["capo_s3.types.grant_read.GrantRead"] = None,
grant_read_acp: Optional["capo_s3.types.grant_read_acp.GrantReadACP"] = None,
grant_write_acp: Optional["capo_s3.types.grant_write_acp.GrantWriteACP"] = None,
write_offset_bytes: Optional[
"capo_s3.types.write_offset_bytes.WriteOffsetBytes"
] = None,
metadata: Optional["capo_s3.types.metadata.Metadata"] = None,
server_side_encryption: Optional[
"capo_s3.types.server_side_encryption.ServerSideEncryption"
] = None,
storage_class: Optional["capo_s3.types.storage_class.StorageClass"] = None,
website_redirect_location: Optional[
"capo_s3.types.website_redirect_location.WebsiteRedirectLocation"
] = None,
sse_customer_algorithm: Optional[
"capo_s3.types.sse_customer_algorithm.SSECustomerAlgorithm"
] = None,
sse_customer_key: Optional[
"capo_s3.types.sse_customer_key.SSECustomerKey"
] = None,
sse_customer_key_md5: Optional[
"capo_s3.types.sse_customer_key_md5.SSECustomerKeyMD5"
] = None,
ssekms_key_id: Optional["capo_s3.types.ssekms_key_id.SSEKMSKeyId"] = None,
ssekms_encryption_context: Optional[
"capo_s3.types.ssekms_encryption_context.SSEKMSEncryptionContext"
] = None,
bucket_key_enabled: Optional[
"capo_s3.types.bucket_key_enabled.BucketKeyEnabled"
] = None,
request_payer: Optional["capo_s3.types.request_payer.RequestPayer"] = None,
tagging: Optional["capo_s3.types.tagging_header.TaggingHeader"] = None,
object_lock_mode: Optional[
"capo_s3.types.object_lock_mode.ObjectLockMode"
] = None,
object_lock_retain_until_date: Optional[
"capo_s3.types.object_lock_retain_until_date.ObjectLockRetainUntilDate"
] = None,
object_lock_legal_hold_status: Optional[
"capo_s3.types.object_lock_legal_hold_status.ObjectLockLegalHoldStatus"
] = None,
object_lock_event_hold: Optional[
"capo_s3.types.object_lock_event_hold.ObjectLockEventHold"
] = None,
object_lock_event_hold_duration_days: Optional[
"capo_s3.types.object_lock_event_hold_duration_days.ObjectLockEventHoldDurationDays"
] = None,
object_lock_event_hold_duration_years: Optional[
"capo_s3.types.object_lock_event_hold_duration_years.ObjectLockEventHoldDurationYears"
] = None,
expected_bucket_owner: Optional["capo_s3.types.account_id.AccountId"] = None,
) -> str:
_, options_ = self.operation_options(config_overrides)
import base64
input_: "capo_s3.types.put_object_request.PutObjectRequest" = {
"bucket": bucket,
"key": key,
"body": ensure_sync_iterator(base64.b64decode("")),
}
if acl is not None:
input_["acl"] = acl
if cache_control is not None:
input_["cache_control"] = cache_control
if content_disposition is not None:
input_["content_disposition"] = content_disposition
if content_encoding is not None:
input_["content_encoding"] = content_encoding
if content_language is not None:
input_["content_language"] = content_language
if content_length is not None:
input_["content_length"] = content_length
if content_md5 is not None:
input_["content_md5"] = content_md5
if content_type is not None:
input_["content_type"] = content_type
if checksum_algorithm is not None:
input_["checksum_algorithm"] = checksum_algorithm
if checksum_crc32 is not None:
input_["checksum_crc32"] = checksum_crc32
if checksum_crc32_c is not None:
input_["checksum_crc32_c"] = checksum_crc32_c
if checksum_crc64_nvme is not None:
input_["checksum_crc64_nvme"] = checksum_crc64_nvme
if checksum_sha1 is not None:
input_["checksum_sha1"] = checksum_sha1
if checksum_sha256 is not None:
input_["checksum_sha256"] = checksum_sha256
if checksum_sha512 is not None:
input_["checksum_sha512"] = checksum_sha512
if checksum_md5 is not None:
input_["checksum_md5"] = checksum_md5
if checksum_xxhash64 is not None:
input_["checksum_xxhash64"] = checksum_xxhash64
if checksum_xxhash3 is not None:
input_["checksum_xxhash3"] = checksum_xxhash3
if checksum_xxhash128 is not None:
input_["checksum_xxhash128"] = checksum_xxhash128
if expires is not None:
input_["expires"] = expires
if if_match is not None:
input_["if_match"] = if_match
if if_none_match is not None:
input_["if_none_match"] = if_none_match
if grant_full_control is not None:
input_["grant_full_control"] = grant_full_control
if grant_read is not None:
input_["grant_read"] = grant_read
if grant_read_acp is not None:
input_["grant_read_acp"] = grant_read_acp
if grant_write_acp is not None:
input_["grant_write_acp"] = grant_write_acp
if write_offset_bytes is not None:
input_["write_offset_bytes"] = write_offset_bytes
if metadata is not None:
input_["metadata"] = metadata
if server_side_encryption is not None:
input_["server_side_encryption"] = server_side_encryption
if storage_class is not None:
input_["storage_class"] = storage_class
if website_redirect_location is not None:
input_["website_redirect_location"] = website_redirect_location
if sse_customer_algorithm is not None:
input_["sse_customer_algorithm"] = sse_customer_algorithm
if sse_customer_key is not None:
input_["sse_customer_key"] = sse_customer_key
if sse_customer_key_md5 is not None:
input_["sse_customer_key_md5"] = sse_customer_key_md5
if ssekms_key_id is not None:
input_["ssekms_key_id"] = ssekms_key_id
if ssekms_encryption_context is not None:
input_["ssekms_encryption_context"] = ssekms_encryption_context
if bucket_key_enabled is not None:
input_["bucket_key_enabled"] = bucket_key_enabled
if request_payer is not None:
input_["request_payer"] = request_payer
if tagging is not None:
input_["tagging"] = tagging
if object_lock_mode is not None:
input_["object_lock_mode"] = object_lock_mode
if object_lock_retain_until_date is not None:
input_["object_lock_retain_until_date"] = object_lock_retain_until_date
if object_lock_legal_hold_status is not None:
input_["object_lock_legal_hold_status"] = object_lock_legal_hold_status
if object_lock_event_hold is not None:
input_["object_lock_event_hold"] = object_lock_event_hold
if object_lock_event_hold_duration_days is not None:
input_["object_lock_event_hold_duration_days"] = (
object_lock_event_hold_duration_days
)
if object_lock_event_hold_duration_years is not None:
input_["object_lock_event_hold_duration_years"] = (
object_lock_event_hold_duration_years
)
if expected_bucket_owner is not None:
input_["expected_bucket_owner"] = expected_bucket_owner
import capo_s3._operations.amazon_s3.put_object
request = capo_s3._operations.amazon_s3.put_object.build_request(
options_, input_
)
request = strip_checksum_headers(request)
signer = (request.context or {}).get("signer")
if not isinstance(signer, SigV4Signer):
raise RuntimeError("presign requires SigV4 credentials")
creds = signer.provider.resolve_identity()
ctx: capo_s3._auth._sigv4.SigV4AuthContext = {
"type": "sig_v4",
"access_key_id": creds["access_key"],
"secret_access_key": creds["secret_key"],
"session_token": creds.get("session_token"),
"signing_region": signer._auth_scheme["signingRegion"],
"signing_name": signer._auth_scheme["signingName"],
"disable_double_encoding": signer._auth_scheme.get(
"disableDoubleEncoding", False
),
"disable_normalize_path": signer._auth_scheme.get(
"disableNormalizePath", False
),
}
return str(presign_sigv4(request, ctx, expires_in=expire_in).url)