FileManager
Once you get the data you can use Katonic Filemanager to access, store and update or manipulate objects within the file manager with Katonic SDK.
From the left panel, Go to File Manager, Click on the Access Token onto the right, Generate Access key and secret key by clicking on Create Access Token.
Install packages
import os
os.system("pip install katonic[filemanager]==1.6.2")
- Create json for filemanager credentials.
filemanager-credentials.json
{
"ACCESS_KEY" : "your access key",
"SECRET_KEY" : "your secret key",
"PRIVATE_BUCKET" : "your private bucket name",
"PUBLIC_BUCKET" : "shared-storage",
}
# loading filemanager credentials
import json
with open('filemanager-credentials.json', 'r') as f:
configs = json.load(f)
Bucket Operations
List Buckets
List information of all accessible buckets.
Return value
A list of buckets.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
buckets = client.list_buckets()
for bucket in buckets:
print(f"{bucket.name} - {bucket.creation_date}")
>>> models - 2022-05-06 03:23:05.545000+00:00
>>> private-storage-6583 - 2022-05-06 03:40:29.458000+00:00
>>> shared-storage - 2022-05-06 03:23:05.492000+00:00
Exists Buckets
Check if a bucket exists.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
if client.bucket_exists("models"):
print("my-bucket exists")
else:
print("my-bucket does not exist")
>>> my-bucket exists
List Objects
Lists object information of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| Prefix | str | Object name start with prefix. |
| Recursive | bool | List recursively than directory structure emulation. |
| start_after | str | List object after this key name. |
| include_user_meta | bool | Flag to control to include user metadata. |
| include_version | bool | Flag to control whether include Object version. |
| use_api_v1 | bool | Flag to control to use ListObjectV1 S3 API or not. |
| use_url_encoding_type | bool | Flag to control whether to use URL encoding type or not. |
Return Value
An iterator of object.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
# List objects information.
objects = client.list_objects(configs['PUBLIC_BUCKET'])
for obj in objects:
print(obj.object_name)
>>> CountVectorizer.pkl
>>> CountVectorizer_1.pkl
>>> TFIDF_transformer.pkl
>>> sample-file.txt
>>> tfidf.pkl
>>> files/
>>> movies_data/
>>> shared-storage /
>>> xyz/
Get Bucket Notification
Get notification configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Return value
NotificationConfig object.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
# Get notification
config = client.get_bucket_notification(configs['PUBLIC_BUCKET'])
Listen Bucket Notification
Listen events of object prefix and suffix of a bucket. Caller should iterate returned iterator to read new events.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| prefix | str | Listen events of objects start with prefix. |
| suffix | str | Listen events of objects ends with suffix. |
| events | list | Events to Listen. |
Return Value
Iterator of event records as dict.
Example
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
with client.listen_bucket_notification(
configs['PUBLIC_BUCKET'],
events = ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
) as events:
for event in events:
print(event)
Get Bucket Encryption
Get encryption configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Return value
SSEConfig object.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
# Get Bucket Encryption
config = client.get_bucket_encryption(configs['PUBLIC_BUCKET'])
Get Bucket Versioning
Get encryption configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
# Get Bucket Versioning
config = client.get_bucket_versioning(configs['PUBLIC_BUCKET'])
print(config.status)
Get Bucket Lifecycle
Get Lifecycle configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Return value
LifecycleConfig object.
Example
# get bucket Lifecycle
client.get_bucket_lifecycle(configs['PUBLIC_BUCKET'])
Get Bucket tags
Get tags configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Return value
tags object.
Example
# get bucket tags
client.get_bucket_tags(configs['PUBLIC_BUCKET'])
Get Object Lock config
Get Object lock configuration of a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
Return value
ObjectLockConfig object.
Example
# get Object lock configuration
config = client.get_object_lock_config(configs['PUBLIC_BUCKET'])
Object Operations
Get Object
Gets data from offset to length of an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| offset | int | Start byte position of object data. |
| length | int | Number of bytes of object data from offset. |
| request_headers | dict | Any additional headers to be added with GET request |
| version_id | str | Version ID of the object. |
| extra_query_params | dict | Extra query parameters for advanced usage. |
Return value
urllib3.response.HTTPResponse object.
Example
# Get data of an object.
try:
response = client.get_object(
configs['PUBLIC_BUCKET'],
"sample-file.txt",
)
# Read data from response.
finally:
response.close()
response.release_conn()
# Get data of an object of version-ID.
try:
response = client.get_object(
configs['PUBLIC_BUCKET'],
"sample-file.txt",
version_id="3081142b-a876-47b3-9cce-13444c78488f",
)
# Read data from response.
finally:
response.close()
response.release_conn()
Select object Content
Select content of an object by SQL expression.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| request | SelectRequest | Select Request. |
Return Value
A reader contains requested records and progress information as SelectObjectReader.
Example
from katonic.filemanager.select import (CSVInputSerialization, CSVOutputSerialization, SelectRequest)
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
with client.select_object_content(
configs['PUBLIC_BUCKET'],
"movies_data/2020.csv",
SelectRequest(
"select * from S3Object",
CSVInputSerialization(),
CSVOutputSerialization(),
request_progress=True,
),
) as result:
for data in result.stream():
print(data.decode())
print(result.stats())
Get file Object
Downloads data of an object to file.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| file_path | str | Name of file to download. |
| request_headers | dict | Any additional headers to be added with GET request |
| version_id | str | Version ID of the object. |
| extra_query_params | dict | Extra query parameters for advanced usage. |
| tmp_file_path | str | Path to a temporary file. |
Return Value
Object information as Object.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
client.fget_object(
configs['PUBLIC_BUCKET'],
"sample-file.txt",
"/kfs_public/files/sample-file.txt"
)
Copy Object
Create an object by server-side copying data from another object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| source | CopySource | Source Object information. |
Return values
ObjectWriteResult object.
Example
from datetime import datetime, timezone
from katonic.filemanager.commonconfig import CopySource
# copy an object from a bucket to another.
result = client.copy_object(
configs['PRIVATE_BUCKET'],
"sample",
CopySource( configs['PUBLIC_BUCKET'], "movies_data/2020.csv"),
)
print(result.object_name, result.version_id)
# copy an object with condition.
result = client.copy_object(
configs['PRIVATE_BUCKET'],
"sample_2",
CopySource(
configs['PUBLIC_BUCKET'], "movies_data/2020.csv",
modified_since=datetime(2014, 4, 1, tzinfo=timezone.utc),
),
)
print(result.object_name, result.version_id)
# copy an object from a bucket with replacing metadata.
metadata = {"test_meta_key": "test_meta_value"}
result = client.copy_object(
configs['PRIVATE_BUCKET'],
"sample_3",
CopySource(configs['PUBLIC_BUCKET'], "movies_data/2020.csv"),
metadata=metadata,
metadata_directive='REPLACE',
)
print(result.object_name, result.version_id)
Compose Object
Create an object by combining data from different source objects using server-side copy.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| source | CopySource | Source Object information. |
Return values
ObjectWriteResult object.
Minimum size required is 5MB
Example
from katonic.filemanager.commonconfig import ComposeSource
sources = [
ComposeSource(configs['PUBLIC_BUCKET'], "movies_data/sample_object_1.csv"),
ComposeSource(configs['PUBLIC_BUCKET'], "movies_data/sample_object_2.csv"),
]
# Create my-bucket/my-object by combining source object
# list.
result = client.compose_object(configs['PUBLIC_BUCKET'], "sample_movies.csv", sources)
print(result.object_name, result.version_id)
# Create my-bucket/my-object with user metadata by combining
# source object list.
result = client.compose_object(configs['PUBLIC_BUCKET'], "sample_movies", sources,metadata={"test_meta_key": "test_meta_value"})
print(result.object_name, result.version_id)
Put file Object
Uploads data from a file to an object in a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| file_path | str | Name of file to upload. |
| content_type | str | Content type of the object. |
| metadata | dict | Any additional metadata to be uploaded along with your PUT request. |
| progress | threading | A progress object. |
| part_size | int | Multipart part size. |
| tags | tags | Tags for the object. |
| retention | Retention | Retention configuration. |
| legal_hold | bool | Flag to set legal hold for the object. |
Return values
ObjectWriteResult object.
Example
from katonic.filemanager.session import Filemanager
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
client.fput_object(
configs['PRIVATE_BUCKET'],
"/kfs_private/",
"/kfs_public/files/sample-file.txt"
)
With Progress Bar
import io
result = fm.put_file_object(
configs['PUBLIC_BUCKET'], "sample_put_object_9.json","sample.json"
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> sample_put_object_9.json: |####################| 0.00 MB/0.00 MB 100% [elapsed: 00:00 left: 00:00, 0.10 MB/sec]created sample_put_object_9.json object; etag: cdde90f633e1fcc0611bd0c3e66318a2, version-id: fed62bfb-4fcf-41ba-8ff4-c974854c0d04
Put Object
Uploads data from a stream to an object in a bucket.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| data | Object | An object having callable read() returning bytes object. |
| length | int | Data size; -1 for unknown size and set valid part_size. |
| content_type | str | Content type of the object. |
| metadata | dict | Any additional metadata to be uploaded along with your PUT request. |
| progress | threading | A progress object. |
| part_size | int | Multipart part size. |
| Tags | tags | Tags for the object. |
| retention | Retention | Retention configuration. |
| legal_hold | bool | Flag to set legal hold for the object.. |
Return values
ObjectWriteResult object.
Example
# Initiating a FileManager Client, so it will connect to our File Manager.
fm = Filemanager(
access_key = configs['ACCESS_KEY'],
secret_key = configs['SECRET_KEY'],
)
client = fm.clientV1
Upload data
import io
result = client.put_object(
configs['PUBLIC_BUCKET'], "sample_put_object", io.BytesIO(b"hello"), 5,
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> created sample_put_object object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: 0898f516-15ba-49c6-92ab-8c93f459aeab
With Progress Bar
# Upload data.
import io
result = fm.put_byte_object(
configs['PUBLIC_BUCKET'], "sample_put_object", io.BytesIO(b"hello"), 5,
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> sample_put_object: |####################| 0.00 MB/0.00 MB 100% [elapsed: 00:00 left: 00:00, 0.00 MB/sec]created sample_put_object object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: 68d67118-842d-47d1-b8a1-ff6d0016ed5b
Upload data with content-type.
import io
result = client.put_object(
configs['PUBLIC_BUCKET'], "sample_put_object_3", io.BytesIO(b"hello"), 5,
content_type="application/csv",
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> created sample_put_object_3 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: e742db64-9295-47af-92cd-7f177ec2c1bf
With Progress Bar
# Upload data with content-type.
import io
result = fm.put_byte_object(
configs['PUBLIC_BUCKET'], "sample_put_object_3", io.BytesIO(b"hello"), 5,
content_type="application/csv",
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> sample_put_object_3: |####################| 0.00 MB/0.00 MB 100% [elapsed: 00:00 left: 00:00, 0.00 MB/sec]created sample_put_object_3 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: f91fc03e-24de-4041-b0a8-8b1767256b7a
Upload data with metadata.
import io
result = client.put_object(
configs['PUBLIC_BUCKET'], "sample_put_object_4", io.BytesIO(b"hello"), 5,
metadata={"My-Project": "one"},
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> created sample_put_object_4 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: 11614dd2-6550-4d7b-b209-65230bbbde53
With Progress Bar
# Upload data with metadata.
import io
result = fm.put_byte_object(
configs['PUBLIC_BUCKET'], "sample_put_object_4", io.BytesIO(b"hello"), 5,
metadata={"My-Project": "one"},
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> sample_put_object_4: |####################| 0.00 MB/0.00 MB 100% [elapsed: 00:00 left: 00:00, 0.00 MB/sec]created sample_put_object_4 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: ea540996-38a4-4b25-9497-d11ce00eca4f
Upload data with tags, retention and legal-hold.
import io
from datetime import datetime, timezone, timedelta
from katonic.filemanager.commonconfig import Tags
from katonic.filemanager.retention import Retention
date = datetime.utcnow().replace(
hour=0, minute=0, second=0, microsecond=0,
) + timedelta(days=30)
tags = Tags(for_object=True)
tags["User"] = "XYZzz"
result = client.put_object(
configs['PUBLIC_BUCKET'], "sample_put_object_8", io.BytesIO(b"hello"), 5,
tags=tags,
retention=Retention('GOVERNANCE', date),
legal_hold=True,
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> created sample_put_object_8 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: bdec99bf-3110-4968-b615-c1153aa50a24
With Progress Bar
# Upload data with tags, retention and legal-hold.
import io
from datetime import datetime, timezone, timedelta
from katonic.filemanager.commonconfig import Tags
from katonic.filemanager.retention import Retention
date = datetime.utcnow().replace(
hour=0, minute=0, second=0, microsecond=0,
) + timedelta(days=30)
tags = Tags(for_object=True)
tags["User"] = "XYZzz"
result = fm.put_byte_object(
configs['PUBLIC_BUCKET'], "sample_put_object_8", io.BytesIO(b"hello"), 5,
tags=tags,
retention=Retention("GOVERNANCE", date),
legal_hold=True,
)
print(
"created {0} object; etag: {1}, version-id: {2}".format(
result.object_name, result.etag, result.version_id,
),
)
>>> sample_put_object_8: |####################| 0.00 MB/0.00 MB 100% [elapsed: 00:00 left: 00:00, 0.00 MB/sec]created sample_put_object_8 object; etag: 5d41402abc4b2a76b9719d911017c592, version-id: ab2494a1-e7bb-4c0b-b348-86285b665843
Remove Object
from katonic.filemanager.deleteobjects import DeleteObject
# Remove list of objects.
errors = client.remove_objects(
"bucket-name",
[
DeleteObject("my-object1"),
DeleteObject("my-object2"),
DeleteObject("my-object3", "622cbc42-f317-41e6-89d5-a0f7ed88eb4d"),
],
)
for error in errors:
print("error occured when deleting object", error)
# Remove a prefix/folder-name recursively.
delete_object_list = map(
lambda x: DeleteObject(x.object_name),
client.list_objects("bucket-name", "folder-name", recursive=True),
)
errors = client.remove_objects("bucket-name", delete_object_list)
for error in errors:
print("error occured when deleting object", error)
Stat Object
Get object information and metadata of an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
| extra_query_params | dict | Extra query parameters for advanced usage. |
Return value
Object information as Object
Example
# Get object information.
result = client.stat_object(configs['PUBLIC_BUCKET'], "sample_put_object_8")
print(
"last-modified: {0}, size: {1}".format(
result.last_modified, result.size,
),
)
>>> last-modified: 2022-05-31 11:36:34+00:00, size: 5
# Get object information of version-ID.
result = client.stat_object(
configs['PUBLIC_BUCKET'], "sample_put_object_8",
version_id="07a31a75-882f-49e7-90c0-dd557514c8f5",
)
print(
"last-modified: {0}, size: {1}".format(
result.last_modified, result.size,
),
)
>>> last-modified: 2022-05-31 11:36:34+00:00, size: 5
Get Object Tags
Get tags configuration of an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
Return value
Tags object.
Example
tags = client.get_object_tags(configs['PUBLIC_BUCKET'], "sample_put_object_8")
========OUTPUT=========
{'User': 'XYZzz'}
Set Object Tags
Set tags configuration to an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
| tags | Tags | Tags configuration. |
Example
# set object tags
tags = Tags.new_object_tags()
tags["User"] = "Disha"
client.set_object_tags(configs['PUBLIC_BUCKET'], "sample_put_object_8", tags)
>>> {'User': 'Disha'}
Enable Object Legal Hold
Enable legal hold on an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
Example
client.enable_object_legal_hold(configs['PUBLIC_BUCKET'], "sample_put_object_8")
Disable Object Legal Hold
Disable legal hold on an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
Example
client.disable_object_legal_hold(configs['PUBLIC_BUCKET'], "sample_put_object_8")
Is object Legal Hold Enabled
Returns true if legal hold is enabled on an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
Example
if client.is_object_legal_hold_enabled(configs['PUBLIC_BUCKET'], "sample_put_object_8"):
print("legal hold is enabled on my-object")
else:
print("legal hold is not enabled on my-object")
>>> legal hold is enabled on my-object
Get Object Retention
Get retention information of an object.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
Return Value
Retention object.
Example
config = client.get_object_retention(configs['PUBLIC_BUCKET'], "sample_put_object_8")
print(config)
Presigned Get Object
Get presigned URL of an object to download its data with expiry time and custom request parameters.
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
| expires | datetime.timedelta | Expiry in seconds; defaults to 7 days. |
| response_headers | dict | Optional response_headers argument to specify response fields like date, size, type of file, data about server, etc. |
| request_date | datetime.timedelta | Optional request_date argument to specify a different request date. Default is current date. |
| extra_query_params | dict | Extra query parameters for advanced usage. |
Return values
URL string
Example
# Get presigned URL string to download 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']' with default expiry (i.e. 7 days).
url = client.presigned_get_object(configs['PUBLIC_BUCKET'], "sample_put_object_8")
print(url)
# Get presigned URL string to download 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']'with two hours expiry.
url = client.presigned_get_object(
configs['PUBLIC_BUCKET'], "sample_put_object_8", expires=timedelta(hours=2),
)
print(url)
Presigned put Object
Get presigned URL of an object to upload data with expiry time and custom request parameters..
Parameters
| Param | type | Description |
|---|---|---|
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| expires | datetime.timedelta | Expiry in seconds; defaults to 7 days. |
Return values
URL string
Example
# Get presigned URL string to upload 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET'] with default expiry (i.e. 7 days).
url = client.presigned_put_object(configs['PUBLIC_BUCKET'], "sample_put_object_8")
print(url)
# Get presigned URL string to uploag 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']'with two hours expiry.
url = client.presigned_put_object(
configs['PUBLIC_BUCKET'], "sample_put_object_8", expires=timedelta(hours=2),
)
print(url)
Get Presigned URL
Get presigned URL of an object for HTTP method, expiry time and custom request parameters.
Parameters
| Param | type | Description |
|---|---|---|
| method | str | HTTP method. |
| bucket_name | str | Name of the bucket. |
| object_name | str | Object Name in the bucket. |
| version_id | str | Version ID of the object. |
| expires | datetime.timedelta | Expiry in seconds; defaults to 7 days. |
| response_headers | dict | Optional response_headers argument to specify response fields like date, size, type of file, data about server, etc. |
| request_date | datetime.timedelta | Optional request_date argument to specify a different request date. Default is current date. |
| extra_query_params | dict | Extra query parameters for advanced usage. |
Return values
URL string
Example
# Get presigned URL string to delete 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']' with one day expiry.
url = client.get_presigned_url(
"DELETE",
configs['PUBLIC_BUCKET'], "sample_put_object_8",
expires=timedelta(days=1),
)
print(url)
# Get presigned URL string to upload 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']' with response-content-type as application/json
# and one day expiry.
url = client.get_presigned_url(
"PUT",
configs['PUBLIC_BUCKET'], "sample_put_object_8",
expires=timedelta(days=1),
response_headers={"response-content-type": "application/json"},
)
print(url)
# Get presigned URL string to download 'sample_put_object_8' in
# 'configs['PUBLIC_BUCKET']' with two hours expiry.
url = client.get_presigned_url(
"GET",
configs['PUBLIC_BUCKET'], "sample_put_object_8",
expires=timedelta(hours=2),
)
print(url)