0
# Configuration Classes
1
2
Configuration objects for bucket settings, server-side encryption, lifecycle management, object policies, versioning, and other advanced S3 features. These classes provide type-safe configuration for sophisticated object storage operations.
3
4
## Capabilities
5
6
### Server-Side Encryption (SSE)
7
8
Configure encryption for objects and buckets with support for customer keys, AWS KMS, and S3-managed encryption.
9
10
```python { .api }
11
class Sse:
12
"""Base class for server-side encryption configurations."""
13
def headers(self) -> dict[str, str]:
14
"""
15
Return HTTP headers for encryption configuration.
16
17
Returns:
18
Dictionary of headers for S3 requests
19
"""
20
21
def tls_required(self) -> bool:
22
"""
23
Check if TLS is required for this encryption method.
24
25
Returns:
26
True if TLS/HTTPS is required
27
"""
28
29
def copy_headers(self) -> dict[str, str]:
30
"""
31
Return HTTP headers for copy operations.
32
33
Returns:
34
Dictionary of headers for S3 copy requests
35
"""
36
37
class SseCustomerKey(Sse):
38
"""Server-side encryption with customer-provided keys (SSE-C)."""
39
def __init__(self, key: bytes) -> None:
40
"""
41
Initialize SSE-C encryption.
42
43
Args:
44
key: 32-byte AES256 encryption key
45
46
Raises:
47
ValueError: If key is not exactly 32 bytes
48
"""
49
50
def headers(self) -> dict[str, str]:
51
"""
52
Return SSE-C headers.
53
54
Returns:
55
Headers with customer encryption key and algorithm
56
"""
57
58
class SseKMS(Sse):
59
"""Server-side encryption with AWS Key Management Service."""
60
def __init__(
61
self,
62
key: str,
63
context: dict[str, str] | None = None
64
) -> None:
65
"""
66
Initialize SSE-KMS encryption.
67
68
Args:
69
key: KMS key ID or ARN
70
context: Encryption context key-value pairs (optional)
71
"""
72
73
def headers(self) -> dict[str, str]:
74
"""
75
Return SSE-KMS headers.
76
77
Returns:
78
Headers with KMS key and encryption context
79
"""
80
81
class SseS3(Sse):
82
"""Server-side encryption with S3-managed keys (SSE-S3)."""
83
def __init__(self) -> None:
84
"""Initialize SSE-S3 encryption with S3-managed keys."""
85
86
def headers(self) -> dict[str, str]:
87
"""
88
Return SSE-S3 headers.
89
90
Returns:
91
Headers for S3-managed encryption
92
"""
93
94
class SSEConfig:
95
"""Bucket default encryption configuration."""
96
def __init__(self, rule: SSERule) -> None:
97
"""
98
Initialize bucket encryption configuration.
99
100
Args:
101
rule: SSE rule defining encryption settings
102
"""
103
rule: SSERule
104
105
@classmethod
106
def fromxml(cls, element: Element) -> SSEConfig:
107
"""Create SSEConfig from XML element."""
108
109
def toxml(self, element: Element) -> Element:
110
"""Convert SSEConfig to XML element."""
111
112
class SSERule:
113
"""Rule for bucket default encryption."""
114
def __init__(
115
self,
116
sse_algorithm: str,
117
kms_master_key_id: str | None = None
118
) -> None:
119
"""
120
Initialize SSE rule.
121
122
Args:
123
sse_algorithm: Encryption algorithm ("AES256" or "aws:kms")
124
kms_master_key_id: KMS key ID (required for aws:kms)
125
"""
126
sse_algorithm: str
127
kms_master_key_id: str | None
128
```
129
130
### Tags and Metadata
131
132
Manage object and bucket tags for organization, billing, and lifecycle management.
133
134
```python { .api }
135
class Tags(dict[str, str]):
136
"""Dictionary extended for bucket/object tags with validation."""
137
def __init__(self, for_object: bool = False) -> None:
138
"""
139
Initialize tags container.
140
141
Args:
142
for_object: True for object tags (max 10), False for bucket tags (max 50)
143
"""
144
145
@classmethod
146
def new_bucket_tags(cls) -> Tags:
147
"""
148
Create new bucket tags instance.
149
150
Returns:
151
Tags instance configured for bucket use (max 50 tags)
152
"""
153
154
@classmethod
155
def new_object_tags(cls) -> Tags:
156
"""
157
Create new object tags instance.
158
159
Returns:
160
Tags instance configured for object use (max 10 tags)
161
"""
162
163
@classmethod
164
def fromxml(cls, element: Element) -> Tags:
165
"""Create Tags from XML element."""
166
167
def toxml(self, element: Element) -> Element:
168
"""Convert Tags to XML element."""
169
```
170
171
### Object Copy and Compose Operations
172
173
Configuration classes for advanced object operations including copying and composing multiple sources.
174
175
```python { .api }
176
class CopySource:
177
"""Copy source specification for copy operations."""
178
def __init__(
179
self,
180
bucket_name: str,
181
object_name: str,
182
version_id: str | None = None,
183
ssec: SseCustomerKey | None = None,
184
offset: int | None = None,
185
length: int | None = None,
186
match_etag: str | None = None,
187
not_match_etag: str | None = None,
188
modified_since: datetime.datetime | None = None,
189
unmodified_since: datetime.datetime | None = None
190
) -> None:
191
"""
192
Initialize copy source.
193
194
Args:
195
bucket_name: Source bucket name
196
object_name: Source object name
197
version_id: Specific version to copy (optional)
198
ssec: Server-side encryption key for encrypted source
199
offset: Byte offset to start copying from
200
length: Number of bytes to copy
201
match_etag: Copy only if ETag matches
202
not_match_etag: Copy only if ETag doesn't match
203
modified_since: Copy only if modified after this date
204
unmodified_since: Copy only if not modified after this date
205
"""
206
bucket_name: str
207
object_name: str
208
version_id: str | None
209
ssec: SseCustomerKey | None
210
offset: int | None
211
length: int | None
212
213
class ComposeSource:
214
"""Compose source specification for compose operations."""
215
def __init__(
216
self,
217
bucket_name: str,
218
object_name: str,
219
version_id: str | None = None,
220
ssec: SseCustomerKey | None = None,
221
offset: int | None = None,
222
length: int | None = None,
223
match_etag: str | None = None,
224
not_match_etag: str | None = None,
225
modified_since: datetime.datetime | None = None,
226
unmodified_since: datetime.datetime | None = None
227
) -> None:
228
"""
229
Initialize compose source.
230
231
Args:
232
bucket_name: Source bucket name
233
object_name: Source object name
234
version_id: Specific version to include (optional)
235
ssec: Server-side encryption key for encrypted source
236
offset: Byte offset to start including from
237
length: Number of bytes to include
238
match_etag: Include only if ETag matches
239
not_match_etag: Include only if ETag doesn't match
240
modified_since: Include only if modified after this date
241
unmodified_since: Include only if not modified after this date
242
"""
243
bucket_name: str
244
object_name: str
245
version_id: str | None
246
ssec: SseCustomerKey | None
247
offset: int | None
248
length: int | None
249
```
250
251
### Versioning Configuration
252
253
Configure bucket versioning to maintain multiple versions of objects.
254
255
```python { .api }
256
class VersioningConfig:
257
"""Bucket versioning configuration."""
258
259
# Constants for versioning status
260
ENABLED: str = "Enabled"
261
SUSPENDED: str = "Suspended"
262
OFF: str = "Off"
263
264
def __init__(
265
self,
266
status: str | None = None,
267
mfa_delete: str | None = None
268
) -> None:
269
"""
270
Initialize versioning configuration.
271
272
Args:
273
status: Versioning status (ENABLED, SUSPENDED, or OFF)
274
mfa_delete: MFA delete requirement (ENABLED or DISABLED)
275
"""
276
status: str | None
277
mfa_delete: str | None
278
279
@classmethod
280
def fromxml(cls, element: Element) -> VersioningConfig:
281
"""Create VersioningConfig from XML element."""
282
283
def toxml(self, element: Element) -> Element:
284
"""Convert VersioningConfig to XML element."""
285
```
286
287
### Lifecycle Management
288
289
Configure automated lifecycle policies for object transitions and expiration.
290
291
```python { .api }
292
class LifecycleConfig:
293
"""Bucket lifecycle configuration with rules."""
294
def __init__(self, rules: list[Rule]) -> None:
295
"""
296
Initialize lifecycle configuration.
297
298
Args:
299
rules: List of lifecycle rules
300
"""
301
rules: list[Rule]
302
303
@classmethod
304
def fromxml(cls, element: Element) -> LifecycleConfig:
305
"""Create LifecycleConfig from XML element."""
306
307
def toxml(self, element: Element) -> Element:
308
"""Convert LifecycleConfig to XML element."""
309
310
class Rule:
311
"""Individual lifecycle rule."""
312
def __init__(
313
self,
314
rule_id: str,
315
status: str,
316
rule_filter: Filter | None = None,
317
expiration: Expiration | None = None,
318
noncurrent_version_expiration: NoncurrentVersionExpiration | None = None,
319
abort_incomplete_multipart_upload: AbortIncompleteMultipartUpload | None = None,
320
transition: Transition | None = None,
321
noncurrent_version_transition: NoncurrentVersionTransition | None = None
322
) -> None:
323
"""
324
Initialize lifecycle rule.
325
326
Args:
327
rule_id: Unique identifier for the rule
328
status: Rule status ("Enabled" or "Disabled")
329
rule_filter: Filter to determine which objects the rule applies to
330
expiration: Object expiration configuration
331
noncurrent_version_expiration: Non-current version expiration
332
abort_incomplete_multipart_upload: Multipart upload cleanup
333
transition: Storage class transition configuration
334
noncurrent_version_transition: Non-current version transitions
335
"""
336
rule_id: str
337
status: str
338
rule_filter: Filter | None
339
expiration: Expiration | None
340
341
class Filter:
342
"""Filter for lifecycle rules."""
343
def __init__(
344
self,
345
prefix: str | None = None,
346
tag: Tag | None = None,
347
and_filter: And | None = None
348
) -> None:
349
"""Initialize lifecycle filter."""
350
prefix: str | None
351
tag: Tag | None
352
and_filter: And | None
353
354
class Expiration:
355
"""Object expiration configuration."""
356
def __init__(
357
self,
358
date: datetime.date | None = None,
359
days: int | None = None,
360
expired_object_delete_marker: bool | None = None
361
) -> None:
362
"""
363
Initialize expiration configuration.
364
365
Args:
366
date: Specific expiration date
367
days: Days after object creation to expire
368
expired_object_delete_marker: Remove expired delete markers
369
"""
370
date: datetime.date | None
371
days: int | None
372
expired_object_delete_marker: bool | None
373
374
class Transition:
375
"""Storage class transition configuration."""
376
def __init__(
377
self,
378
date: datetime.date | None = None,
379
days: int | None = None,
380
storage_class: str | None = None
381
) -> None:
382
"""
383
Initialize transition configuration.
384
385
Args:
386
date: Specific transition date
387
days: Days after object creation to transition
388
storage_class: Target storage class (GLACIER, DEEP_ARCHIVE, etc.)
389
"""
390
date: datetime.date | None
391
days: int | None
392
storage_class: str | None
393
```
394
395
### Bucket Notifications
396
397
Configure bucket notifications for real-time event handling.
398
399
```python { .api }
400
class NotificationConfig:
401
"""Bucket notification configuration."""
402
def __init__(
403
self,
404
cloud_func_config_list: list[CloudFunctionConfig] | None = None,
405
queue_config_list: list[QueueConfig] | None = None,
406
topic_config_list: list[TopicConfig] | None = None
407
) -> None:
408
"""
409
Initialize notification configuration.
410
411
Args:
412
cloud_func_config_list: Lambda function configurations
413
queue_config_list: SQS queue configurations
414
topic_config_list: SNS topic configurations
415
"""
416
cloud_func_config_list: list[CloudFunctionConfig] | None
417
queue_config_list: list[QueueConfig] | None
418
topic_config_list: list[TopicConfig] | None
419
420
class CloudFunctionConfig:
421
"""Lambda function notification configuration."""
422
def __init__(
423
self,
424
id_: str,
425
cloud_func: str,
426
events: list[str],
427
prefix_filter_rule: PrefixFilterRule | None = None,
428
suffix_filter_rule: SuffixFilterRule | None = None
429
) -> None:
430
"""
431
Initialize Lambda function configuration.
432
433
Args:
434
id_: Unique configuration identifier
435
cloud_func: Lambda function ARN
436
events: List of S3 events to trigger on
437
prefix_filter_rule: Object key prefix filter
438
suffix_filter_rule: Object key suffix filter
439
"""
440
441
class QueueConfig:
442
"""SQS queue notification configuration."""
443
def __init__(
444
self,
445
id_: str,
446
queue: str,
447
events: list[str],
448
prefix_filter_rule: PrefixFilterRule | None = None,
449
suffix_filter_rule: SuffixFilterRule | None = None
450
) -> None:
451
"""Initialize SQS queue configuration."""
452
453
class TopicConfig:
454
"""SNS topic notification configuration."""
455
def __init__(
456
self,
457
id_: str,
458
topic: str,
459
events: list[str],
460
prefix_filter_rule: PrefixFilterRule | None = None,
461
suffix_filter_rule: SuffixFilterRule | None = None
462
) -> None:
463
"""Initialize SNS topic configuration."""
464
```
465
466
### Object Lock and Retention
467
468
Configure object lock for compliance and legal hold requirements.
469
470
```python { .api }
471
class ObjectLockConfig:
472
"""Object lock configuration for compliance."""
473
def __init__(
474
self,
475
retention_mode: str | None = None,
476
retention_duration_days: int | None = None,
477
retention_duration_years: int | None = None
478
) -> None:
479
"""
480
Initialize object lock configuration.
481
482
Args:
483
retention_mode: Retention mode ("GOVERNANCE" or "COMPLIANCE")
484
retention_duration_days: Retention period in days
485
retention_duration_years: Retention period in years
486
"""
487
retention_mode: str | None
488
retention_duration_days: int | None
489
retention_duration_years: int | None
490
491
class Retention:
492
"""Object retention configuration."""
493
def __init__(
494
self,
495
mode: str,
496
retain_until_date: datetime.datetime
497
) -> None:
498
"""
499
Initialize retention configuration.
500
501
Args:
502
mode: Retention mode ("GOVERNANCE" or "COMPLIANCE")
503
retain_until_date: Date until which object is retained
504
"""
505
mode: str
506
retain_until_date: datetime.datetime
507
508
@classmethod
509
def fromxml(cls, element: Element) -> Retention:
510
"""Create Retention from XML element."""
511
512
def toxml(self, element: Element) -> Element:
513
"""Convert Retention to XML element."""
514
515
class LegalHold:
516
"""Legal hold configuration for objects."""
517
def __init__(self, status: bool) -> None:
518
"""
519
Initialize legal hold configuration.
520
521
Args:
522
status: True to enable legal hold, False to disable
523
"""
524
status: bool
525
526
@classmethod
527
def fromxml(cls, element: Element) -> LegalHold:
528
"""Create LegalHold from XML element."""
529
530
def toxml(self, element: Element) -> Element:
531
"""Convert LegalHold to XML element."""
532
```
533
534
### Replication Configuration
535
536
Configure cross-region replication for disaster recovery and compliance.
537
538
```python { .api }
539
class ReplicationConfig:
540
"""Bucket replication configuration."""
541
def __init__(self, role: str, rules: list[ReplicationRule]) -> None:
542
"""
543
Initialize replication configuration.
544
545
Args:
546
role: IAM role ARN for replication
547
rules: List of replication rules
548
"""
549
role: str
550
rules: list[ReplicationRule]
551
552
class ReplicationRule:
553
"""Individual replication rule."""
554
def __init__(
555
self,
556
rule_id: str,
557
status: str,
558
priority: int,
559
delete_marker_replication: DeleteMarkerReplication | None = None,
560
rule_filter: Filter | None = None,
561
destination: Destination | None = None
562
) -> None:
563
"""
564
Initialize replication rule.
565
566
Args:
567
rule_id: Unique identifier for the rule
568
status: Rule status ("Enabled" or "Disabled")
569
priority: Rule priority (higher numbers have precedence)
570
delete_marker_replication: Delete marker replication settings
571
rule_filter: Filter to determine which objects to replicate
572
destination: Replication destination configuration
573
"""
574
rule_id: str
575
status: str
576
priority: int
577
destination: Destination | None
578
579
class Destination:
580
"""Replication destination configuration."""
581
def __init__(
582
self,
583
bucket_arn: str,
584
storage_class: str | None = None,
585
access_control_translation: AccessControlTranslation | None = None,
586
encryption_configuration: EncryptionConfiguration | None = None
587
) -> None:
588
"""
589
Initialize replication destination.
590
591
Args:
592
bucket_arn: Destination bucket ARN
593
storage_class: Target storage class
594
access_control_translation: Access control settings
595
encryption_configuration: Encryption settings for destination
596
"""
597
bucket_arn: str
598
storage_class: str | None
599
```
600
601
## Usage Examples
602
603
### Server-Side Encryption
604
605
```python
606
from minio import Minio
607
from minio.sse import SseCustomerKey, SseKMS, SseS3
608
import secrets
609
610
client = Minio("localhost:9000", "minio", "minio123")
611
612
# Customer-provided key encryption (SSE-C)
613
customer_key = secrets.token_bytes(32) # 32-byte AES key
614
sse_c = SseCustomerKey(customer_key)
615
616
client.put_object(
617
"my-bucket",
618
"encrypted-file.txt",
619
data=io.BytesIO(b"sensitive data"),
620
length=13,
621
sse=sse_c
622
)
623
624
# KMS encryption (SSE-KMS)
625
sse_kms = SseKMS("arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")
626
627
client.put_object(
628
"my-bucket",
629
"kms-encrypted.txt",
630
data=io.BytesIO(b"kms protected data"),
631
length=18,
632
sse=sse_kms
633
)
634
635
# S3-managed encryption (SSE-S3)
636
sse_s3 = SseS3()
637
638
client.put_object(
639
"my-bucket",
640
"s3-encrypted.txt",
641
data=io.BytesIO(b"s3 protected data"),
642
length=17,
643
sse=sse_s3
644
)
645
```
646
647
### Tags and Metadata
648
649
```python
650
from minio.commonconfig import Tags
651
652
# Create bucket tags
653
bucket_tags = Tags.new_bucket_tags()
654
bucket_tags["Environment"] = "Production"
655
bucket_tags["Team"] = "Engineering"
656
bucket_tags["Cost-Center"] = "Engineering-2024"
657
658
client.set_bucket_tags("my-bucket", bucket_tags)
659
660
# Create object tags
661
object_tags = Tags.new_object_tags()
662
object_tags["Type"] = "Document"
663
object_tags["Classification"] = "Internal"
664
665
client.put_object(
666
"my-bucket",
667
"document.pdf",
668
data=io.BytesIO(b"document content"),
669
length=16,
670
tags=object_tags
671
)
672
673
# Retrieve and modify tags
674
existing_tags = client.get_bucket_tags("my-bucket")
675
if existing_tags:
676
existing_tags["LastModified"] = datetime.datetime.now().isoformat()
677
client.set_bucket_tags("my-bucket", existing_tags)
678
```
679
680
### Bucket Versioning
681
682
```python
683
from minio.versioningconfig import VersioningConfig
684
685
# Enable versioning
686
versioning_config = VersioningConfig(VersioningConfig.ENABLED)
687
client.set_bucket_versioning("my-bucket", versioning_config)
688
689
# Check versioning status
690
current_versioning = client.get_bucket_versioning("my-bucket")
691
print(f"Versioning status: {current_versioning.status}")
692
693
# Suspend versioning (keeps existing versions)
694
suspend_config = VersioningConfig(VersioningConfig.SUSPENDED)
695
client.set_bucket_versioning("my-bucket", suspend_config)
696
```
697
698
### Lifecycle Management
699
700
```python
701
from minio.lifecycleconfig import (
702
LifecycleConfig, Rule, Filter, Expiration, Transition
703
)
704
import datetime
705
706
# Create lifecycle rules
707
rules = []
708
709
# Rule 1: Delete objects after 365 days
710
expiration_rule = Rule(
711
rule_id="delete-old-objects",
712
status="Enabled",
713
rule_filter=Filter(prefix="logs/"),
714
expiration=Expiration(days=365)
715
)
716
rules.append(expiration_rule)
717
718
# Rule 2: Transition to Glacier after 30 days
719
transition_rule = Rule(
720
rule_id="archive-documents",
721
status="Enabled",
722
rule_filter=Filter(prefix="documents/"),
723
transition=Transition(days=30, storage_class="GLACIER")
724
)
725
rules.append(transition_rule)
726
727
# Apply lifecycle configuration
728
lifecycle_config = LifecycleConfig(rules)
729
client.set_bucket_lifecycle("my-bucket", lifecycle_config)
730
731
# Get current lifecycle configuration
732
current_lifecycle = client.get_bucket_lifecycle("my-bucket")
733
for rule in current_lifecycle.rules:
734
print(f"Rule: {rule.rule_id}, Status: {rule.status}")
735
```
736
737
### Object Copy with Advanced Options
738
739
```python
740
from minio.commonconfig import CopySource, Tags
741
from minio.sse import SseS3
742
import datetime
743
744
# Copy with conditions and encryption
745
source = CopySource(
746
bucket_name="source-bucket",
747
object_name="source-object.txt",
748
modified_since=datetime.datetime(2024, 1, 1) # Only copy if modified after date
749
)
750
751
# Copy with new metadata and encryption
752
new_tags = Tags.new_object_tags()
753
new_tags["Source"] = "Migrated"
754
new_tags["Date"] = datetime.datetime.now().isoformat()
755
756
result = client.copy_object(
757
bucket_name="dest-bucket",
758
object_name="dest-object.txt",
759
source=source,
760
sse=SseS3(),
761
metadata={"Content-Type": "text/plain", "Author": "System"},
762
tags=new_tags
763
)
764
765
print(f"Copied object ETag: {result.etag}")
766
```
767
768
### Object Retention and Legal Hold
769
770
```python
771
from minio.retention import Retention
772
from minio.legalhold import LegalHold
773
import datetime
774
775
# Set retention policy (WORM - Write Once Read Many)
776
retention_date = datetime.datetime.utcnow() + datetime.timedelta(days=2555) # 7 years
777
retention = Retention("COMPLIANCE", retention_date)
778
779
# Upload with retention
780
client.put_object(
781
"compliance-bucket",
782
"compliance-document.pdf",
783
data=io.BytesIO(b"compliance document"),
784
length=19,
785
retention=retention,
786
legal_hold=True
787
)
788
789
# Modify legal hold status
790
legal_hold = LegalHold(False) # Remove legal hold
791
client.set_object_legal_hold(
792
"compliance-bucket",
793
"compliance-document.pdf",
794
legal_hold
795
)
796
```
797
798
### Bucket Default Encryption
799
800
```python
801
from minio.sseconfig import SSEConfig, SSERule
802
803
# Set bucket default encryption to SSE-S3
804
sse_rule = SSERule("AES256")
805
sse_config = SSEConfig(sse_rule)
806
client.set_bucket_encryption("my-bucket", sse_config)
807
808
# Set bucket default encryption to KMS
809
kms_rule = SSERule("aws:kms", "arn:aws:kms:us-east-1:123456789012:key/12345678-1234-1234-1234-123456789012")
810
kms_config = SSEConfig(kms_rule)
811
client.set_bucket_encryption("kms-bucket", kms_config)
812
813
# Get current encryption configuration
814
current_encryption = client.get_bucket_encryption("my-bucket")
815
if current_encryption:
816
print(f"Encryption algorithm: {current_encryption.rule.sse_algorithm}")
817
```