دليل AWS S3 الكامل 2026: الدلاء، Boto3 وأنماط الإنتاج

# AWS CLI setup
pip install awscli
aws configure  # enter Access Key, Secret, region

# Create bucket
aws s3 mb s3://my-bucket-2026 --region us-east-1

# Upload file
aws s3 cp file.jpg s3://my-bucket-2026/images/file.jpg

# Upload directory
aws s3 sync ./dist s3://my-bucket-2026/website/ --delete

# List objects
aws s3 ls s3://my-bucket-2026/
aws s3 ls s3://my-bucket-2026/images/ --recursive

# Download
aws s3 cp s3://my-bucket-2026/data.csv ./data.csv

# Delete
aws s3 rm s3://my-bucket-2026/old-file.txt
aws s3 rm s3://my-bucket-2026/old-folder/ --recursive

import boto3
from botocore.exceptions import ClientError
from pathlib import Path
import json

# Initialize client
s3 = boto3.client('s3', region_name='us-east-1')

# Or use resource (higher-level API)
s3_resource = boto3.resource('s3')
bucket = s3_resource.Bucket('my-bucket-2026')

# Create bucket
def create_bucket(bucket_name: str, region: str = 'us-east-1') -> bool:
    try:
        if region == 'us-east-1':
            s3.create_bucket(Bucket=bucket_name)
        else:
            s3.create_bucket(
                Bucket=bucket_name,
                CreateBucketConfiguration={'LocationConstraint': region}
            )
        return True
    except ClientError as e:
        print(f"Error: {e}")
        return False

# Upload file
def upload_file(file_path: str, bucket: str, key: str = None) -> bool:
    if key is None:
        key = Path(file_path).name
    try:
        s3.upload_file(file_path, bucket, key,
                       ExtraArgs={'ContentType': 'image/jpeg'})
        return True
    except ClientError as e:
        print(f"Upload failed: {e}")
        return False

# Upload from memory (no temp file needed)
def upload_bytes(data: bytes, bucket: str, key: str, content_type: str = 'application/octet-stream'):
    s3.put_object(Body=data, Bucket=bucket, Key=key, ContentType=content_type)

# Download
def download_file(bucket: str, key: str, dest_path: str):
    s3.download_file(bucket, key, dest_path)

def download_bytes(bucket: str, key: str) -> bytes:
    response = s3.get_object(Bucket=bucket, Key=key)
    return response['Body'].read()

# List objects
def list_objects(bucket: str, prefix: str = '') -> list[dict]:
    paginator = s3.get_paginator('list_objects_v2')
    objects = []
    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
        objects.extend(page.get('Contents', []))
    return objects

# Delete
def delete_object(bucket: str, key: str):
    s3.delete_object(Bucket=bucket, Key=key)

# Enable static hosting
aws s3 website s3://my-website-bucket/   --index-document index.html   --error-document 404.html

# Bucket policy for public access
cat > policy.json << 'EOF'
{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": "*",
    "Action": "s3:GetObject",
    "Resource": "arn:aws:s3:::my-website-bucket/*"
  }]
}
EOF

aws s3api put-bucket-policy   --bucket my-website-bucket   --policy file://policy.json

# Deploy website
aws s3 sync ./dist s3://my-website-bucket/   --delete   --cache-control "max-age=31536000"   --exclude "*.html"

# HTML files: no cache
aws s3 sync ./dist s3://my-website-bucket/   --exclude "*" --include "*.html"   --cache-control "no-cache"

# Website URL: http://my-website-bucket.s3-website-us-east-1.amazonaws.com

# Generate time-limited URLs for private objects

def get_presigned_url(bucket: str, key: str, expiry_seconds: int = 3600) -> str:
    url = s3.generate_presigned_url(
        'get_object',
        Params={'Bucket': bucket, 'Key': key},
        ExpiresIn=expiry_seconds
    )
    return url

# Upload pre-signed URL — let clients upload directly to S3
def get_presigned_upload_url(bucket: str, key: str, content_type: str, expiry: int = 300) -> dict:
    response = s3.generate_presigned_post(
        bucket,
        key,
        Fields={'Content-Type': content_type},
        Conditions=[
            {'Content-Type': content_type},
            ['content-length-range', 1, 10 * 1024 * 1024]  # max 10MB
        ],
        ExpiresIn=expiry
    )
    return response

# FastAPI endpoint returning upload URL
# @app.get('/upload-url')
# def get_upload_url(filename: str, content_type: str):
#     key = f'uploads/{uuid4()}/{filename}'
#     url_data = get_presigned_upload_url(BUCKET, key, content_type)
#     return {'url': url_data['url'], 'fields': url_data['fields'], 'key': key}

{
  "Rules": [{
    "ID": "archive-old-logs",
    "Status": "Enabled",
    "Filter": {"Prefix": "logs/"},
    "Transitions": [
      {"Days": 30, "StorageClass": "STANDARD_IA"},
      {"Days": 90, "StorageClass": "GLACIER"}
    ],
    "Expiration": {"Days": 365}
  }]
}

# Enable versioning
aws s3api put-bucket-versioning   --bucket my-bucket   --versioning-configuration Status=Enabled

# Cross-region replication (in AWS Console or CloudFormation)
# Requires versioning on both source and destination buckets

# List object versions
aws s3api list-object-versions --bucket my-bucket --prefix data.csv