Skip to main content

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.

    Untitled

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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName of the bucket.
PrefixstrObject name start with prefix.
RecursiveboolList recursively than directory structure emulation.
start_afterstrList object after this key name.
include_user_metaboolFlag to control to include user metadata.
include_versionboolFlag to control whether include Object version.
use_api_v1boolFlag to control to use ListObjectV1 S3 API or not.
use_url_encoding_typeboolFlag 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName of the bucket.
prefixstrListen events of objects start with prefix.
suffixstrListen events of objects ends with suffix.
eventslistEvents 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
offsetintStart byte position of object data.
lengthintNumber of bytes of object data from offset.
request_headersdictAny additional headers to be added with GET request
version_idstrVersion ID of the object.
extra_query_paramsdictExtra 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
requestSelectRequestSelect 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
file_pathstrName of file to download.
request_headersdictAny additional headers to be added with GET request
version_idstrVersion ID of the object.
extra_query_paramsdictExtra query parameters for advanced usage.
tmp_file_pathstrPath 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
sourceCopySourceSource 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
sourceCopySourceSource 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
file_pathstrName of file to upload.
content_typestrContent type of the object.
metadatadictAny additional metadata to be uploaded along with your PUT request.
progressthreadingA progress object.
part_sizeintMultipart part size.
tagstagsTags for the object.
retentionRetentionRetention configuration.
legal_holdboolFlag 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
dataObjectAn object having callable read() returning bytes object.
lengthintData size; -1 for unknown size and set valid part_size.
content_typestrContent type of the object.
metadatadictAny additional metadata to be uploaded along with your PUT request.
progressthreadingA progress object.
part_sizeintMultipart part size.
TagstagsTags for the object.
retentionRetentionRetention configuration.
legal_holdboolFlag 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
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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion ID of the object.
extra_query_paramsdictExtra 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion ID of the object.
tagsTagsTags 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 legal hold on an object.

Parameters

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion ID of the object.

Example

client.disable_object_legal_hold(configs['PUBLIC_BUCKET'], "sample_put_object_8")

Returns true if legal hold is enabled on an object.

Parameters

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion ID of the object.
expiresdatetime.timedeltaExpiry in seconds; defaults to 7 days.
response_headersdictOptional response_headers argument to specify response fields like date, size, type of file, data about server, etc.
request_datedatetime.timedeltaOptional request_date argument to specify a different request date. Default is current date.
extra_query_paramsdictExtra 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

ParamtypeDescription
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
expiresdatetime.timedeltaExpiry 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

ParamtypeDescription
methodstrHTTP method.
bucket_namestrName of the bucket.
object_namestrObject Name in the bucket.
version_idstrVersion ID of the object.
expiresdatetime.timedeltaExpiry in seconds; defaults to 7 days.
response_headersdictOptional response_headers argument to specify response fields like date, size, type of file, data about server, etc.
request_datedatetime.timedeltaOptional request_date argument to specify a different request date. Default is current date.
extra_query_paramsdictExtra 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)