Spis treści 12 sekcji
  1. Dlaczego Python + Boto3?
  2. Skrypt 1: Lista wszystkich zasobów z tagami
  3. Skrypt 2: Automatyczne snapshoty EBS
  4. Skrypt 3: Raport kosztów AWS
  5. Skrypt 4: Cleanup nieużywanych zasobów
  6. Skrypt 5: Bulk tagging zasobów
  7. Skrypt 6: S3 bucket analyzer
  8. Skrypt 7: Security Group audit
  9. Skrypt 8: Lambda - auto-stop instancji dev wieczorem
  10. Uruchamianie skryptów jako Lambda + EventBridge
  11. FAQ - najczęstsze pytania o Boto3
  12. Co dalej?
TL;DR: Boto3 to oficjalna biblioteka Python do AWS. Pozwala programistycznie zarządzać dowolnym serwisem AWS - od tworzenia EC2 przez zarządzanie S3 po automatyzację backupów. W tym artykule pokażę 8 praktycznych skryptów, które oszczędzają godziny ręcznej pracy: auto-snapshot, cleanup nieużywanych zasobów, raport kosztów, tagging audit i więcej.

Dlaczego Python + Boto3?

Klikanie w konsoli AWS jest OK na początku. Ale gdy musisz powtórzyć tę samą operację na 20 instancjach, stworzyć raport kosztów co tydzień, albo automatycznie czyścić stare snapshoty - ręczna praca nie ma sensu.

Boto3 to oficjalna biblioteka AWS SDK dla Python. Daje dostęp do każdego serwisu AWS przez prosty kod Python. Jeśli da się to zrobić w konsoli lub CLI - da się to zautomatyzować z Boto3.

Instalacja i konfiguracja

# Instalacja
pip install boto3

# Konfiguracja credentials (jednorazowo)
aws configure
# AWS Access Key ID: AKIA...
# AWS Secret Access Key: ...
# Default region: eu-central-1
# Default output format: json
Bezpieczeństwo: Na EC2 i Lambda NIE używaj access keys! Przypisz IAM Role - Boto3 automatycznie pobierze credentials. Access keys tylko do lokalnego developmentu.

Pierwsze kroki z Boto3

import boto3

# Client - niskopoziomowy, zwraca dict
ec2_client = boto3.client('ec2', region_name='eu-central-1')
response = ec2_client.describe_instances()

# Resource - wysokopoziomowy, zwraca obiekty
ec2_resource = boto3.resource('ec2', region_name='eu-central-1')
for instance in ec2_resource.instances.all():
    print(f"{instance.id}: {instance.state['name']}")

# Session - dla wielu regionów/kont
session = boto3.Session(profile_name='produkcja', region_name='eu-central-1')
s3 = session.client('s3')

Skrypt 1: Lista wszystkich zasobów z tagami

Audyt tagowania - znajdź zasoby bez obowiązkowych tagów:

import boto3

ec2 = boto3.resource('ec2', region_name='eu-central-1')
REQUIRED_TAGS = ['Environment', 'Project', 'Owner']

print("=== INSTANCJE BEZ WYMAGANYCH TAGÓW ===")
for instance in ec2.instances.all():
    tags = {t['Key']: t['Value'] for t in (instance.tags or [])}
    missing = [t for t in REQUIRED_TAGS if t not in tags]

    if missing:
        name = tags.get('Name', 'brak nazwy')
        print(f"  {instance.id} ({name}) - brakuje: {', '.join(missing)}")

print("\n=== WOLUMENY BEZ TAGÓW ===")
for vol in ec2.volumes.all():
    tags = {t['Key']: t['Value'] for t in (vol.tags or [])}
    if not tags.get('Project'):
        state = vol.state
        size = vol.size
        attached = vol.attachments[0]['InstanceId'] if vol.attachments else 'nieprzypisany'
        print(f"  {vol.id} - {size}GB, stan: {state}, instancja: {attached}")

Dlaczego tagowanie jest ważne - opisałem w artykule o 10 błędach na AWS.

Skrypt 2: Automatyczne snapshoty EBS

Codzienne backupy wolumenów EBS z automatycznym usuwaniem starych:

import boto3
from datetime import datetime, timedelta

ec2 = boto3.client('ec2', region_name='eu-central-1')
RETENTION_DAYS = 7

def create_snapshots():
    """Utwórz snapshoty wolumenów z tagiem Backup=true"""
    volumes = ec2.describe_volumes(
        Filters=[{'Name': 'tag:Backup', 'Values': ['true']}]
    )['Volumes']

    for vol in volumes:
        tags = {t['Key']: t['Value'] for t in vol.get('Tags', [])}
        name = tags.get('Name', vol['VolumeId'])

        snapshot = ec2.create_snapshot(
            VolumeId=vol['VolumeId'],
            Description=f"Auto backup: {name} - {datetime.now().strftime('%Y-%m-%d')}"
        )

        ec2.create_tags(
            Resources=[snapshot['SnapshotId']],
            Tags=[
                {'Key': 'Name', 'Value': f"backup-{name}-{datetime.now().strftime('%Y%m%d')}" },
                {'Key': 'AutoBackup', 'Value': 'true'},
                {'Key': 'DeleteAfter', 'Value': (datetime.now() + timedelta(days=RETENTION_DAYS)).strftime('%Y-%m-%d')}
            ]
        )
        print(f"Snapshot utworzony: {snapshot['SnapshotId']} dla {name}")

def cleanup_old_snapshots():
    """Usuń snapshoty starsze niż RETENTION_DAYS"""
    snapshots = ec2.describe_snapshots(
        OwnerIds=['self'],
        Filters=[{'Name': 'tag:AutoBackup', 'Values': ['true']}]
    )['Snapshots']

    today = datetime.now().strftime('%Y-%m-%d')
    for snap in snapshots:
        tags = {t['Key']: t['Value'] for t in snap.get('Tags', [])}
        delete_after = tags.get('DeleteAfter', '')

        if delete_after and delete_after <= today:
            ec2.delete_snapshot(SnapshotId=snap['SnapshotId'])
            print(f"Snapshot usunięty: {snap['SnapshotId']} (expired: {delete_after})")

if __name__ == '__main__':
    create_snapshots()
    cleanup_old_snapshots()
Pro tip: Uruchom ten skrypt jako Lambda triggerowana przez EventBridge cron (rate(1 day)). Zero serwerów, zero pamiętania o backupach.

Skrypt 3: Raport kosztów AWS

Cotygodniowy raport kosztów per serwis, wysyłany emailem przez SES:

Koszt API: Cost Explorer API jest płatne - $0.01 za każde zapytanie (get_cost_and_usage, get_cost_forecast). Przy raporcie raz w tygodniu to grosze, ale nie wywołuj tych funkcji w pętli ani przy każdym żądaniu aplikacji.
import boto3
from datetime import datetime, timedelta

ce = boto3.client('ce', region_name='us-east-1')  # Cost Explorer - TYLKO us-east-1

def get_cost_report():
    end = datetime.now().strftime('%Y-%m-%d')
    start = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')

    # Koszty per serwis
    response = ce.get_cost_and_usage(
        TimePeriod={'Start': start, 'End': end},
        Granularity='MONTHLY',
        Metrics=['BlendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )

    report = "=== RAPORT KOSZTÓW AWS ===\n"
    report += f"Okres: {start} do {end}\n\n"

    total = 0
    services = []
    for group in response['ResultsByTime'][0]['Groups']:
        service = group['Keys'][0]
        cost = float(group['Metrics']['BlendedCost']['Amount'])
        if cost > 0.01:
            services.append((service, cost))
            total += cost

    services.sort(key=lambda x: x[1], reverse=True)

    for service, cost in services:
        pln = cost * 4.0  # przybliżony kurs USD/PLN
        report += f"  {service:<40} ${cost:>8.2f}  (~{pln:.0f} zł)\n"

    separator = "=" * 60
    report += f"\n{separator}\n"
    report += f"  SUMA: ${total:.2f} (~{total*4:.0f} zł)\n"

    return report

def get_forecast():
    """Prognoza kosztów na koniec miesiąca"""
    start = datetime.now().strftime('%Y-%m-%d')
    # Koniec miesiąca
    if datetime.now().month == 12:
        end = f"{datetime.now().year + 1}-01-01"
    else:
        end = f"{datetime.now().year}-{datetime.now().month + 1:02d}-01"

    try:
        forecast = ce.get_cost_forecast(
            TimePeriod={'Start': start, 'End': end},
            Metric='BLENDED_COST',
            Granularity='MONTHLY'
        )
        amount = float(forecast['Total']['Amount'])
        return f"\nPrognoza na koniec miesiąca: ${amount:.2f} (~{amount*4:.0f} zł)"
    except Exception:
        return ""

if __name__ == '__main__':
    report = get_cost_report()
    report += get_forecast()
    print(report)

    # Opcjonalnie: wyślij emailem przez SES
    # ses = boto3.client('ses', region_name='eu-central-1')
    # ses.send_email(
    #     Source='[email protected]',
    #     Destination={'ToAddresses': ['[email protected]']},
    #     Message={
    #         'Subject': {'Data': f'AWS Cost Report - {datetime.now().strftime("%B %Y")}'},
    #         'Body': {'Text': {'Data': report}}
    #     }
    # )

Skrypt 4: Cleanup nieużywanych zasobów

Znajdź i (opcjonalnie) usuń zasoby, które generują koszty bez pożytku:

import boto3

ec2 = boto3.client('ec2', region_name='eu-central-1')

def find_unused_ebs_volumes():
    """Wolumeny EBS w stanie 'available' (nieprzypisane do żadnej instancji)"""
    volumes = ec2.describe_volumes(
        Filters=[{'Name': 'status', 'Values': ['available']}]
    )['Volumes']

    total_cost = 0
    print(f"\n=== NIEPRZYPISANE WOLUMENY EBS ({len(volumes)}) ===")
    for vol in volumes:
        cost = vol['Size'] * 0.08  # $0.08/GB/mc dla gp3
        total_cost += cost
        print(f"  {vol['VolumeId']} - {vol['Size']}GB ({vol['VolumeType']}) - ~${cost:.2f}/mc")

    print(f"  POTENCJALNA OSZCZĘDNOŚĆ: ~${total_cost:.2f}/mc")
    return volumes

def find_unused_elastic_ips():
    """Elastic IPs nieprzypisane do instancji (od 2024 każdy publiczny IPv4 kosztuje ~$3.65/mc, nieużywany to czysta strata)"""
    addresses = ec2.describe_addresses()['Addresses']
    unused = [a for a in addresses if 'InstanceId' not in a and 'NetworkInterfaceId' not in a]

    print(f"\n=== NIEPRZYPISANE ELASTIC IPs ({len(unused)}) ===")
    for addr in unused:
        print(f"  {addr['PublicIp']} ({addr['AllocationId']}) - $3.65/mc")

    if unused:
        print(f"  POTENCJALNA OSZCZĘDNOŚĆ: ~${len(unused) * 3.65:.2f}/mc")
    return unused

def find_old_snapshots(days=90):
    """Snapshoty starsze niż N dni"""
    from datetime import datetime, timedelta, timezone
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)

    snapshots = ec2.describe_snapshots(OwnerIds=['self'])['Snapshots']
    old = [s for s in snapshots if s['StartTime'] < cutoff]

    total_size = sum(s['VolumeSize'] for s in old)
    print(f"\n=== SNAPSHOTY STARSZE NIŻ {days} DNI ({len(old)}) ===")
    print(f"  Łączny rozmiar: {total_size}GB (~${total_size * 0.05:.2f}/mc)")
    for s in old[:10]:
        print(f"  {s['SnapshotId']} - {s['VolumeSize']}GB - {s['StartTime'].strftime('%Y-%m-%d')}")
    if len(old) > 10:
        print(f"  ... i {len(old) - 10} więcej")

def find_stopped_instances():
    """Instancje EC2 w stanie 'stopped' (EBS wciąż kosztuje!)"""
    instances = ec2.describe_instances(
        Filters=[{'Name': 'instance-state-name', 'Values': ['stopped']}]
    )

    stopped = []
    for res in instances['Reservations']:
        for inst in res['Instances']:
            stopped.append(inst)

    print(f"\n=== ZATRZYMANE INSTANCJE EC2 ({len(stopped)}) ===")
    for inst in stopped:
        tags = {t['Key']: t['Value'] for t in inst.get('Tags', [])}
        name = tags.get('Name', 'brak nazwy')
        ebs_size = sum(b['Ebs']['VolumeSize'] for b in inst.get('BlockDeviceMappings', []) if 'Ebs' in b)
        print(f"  {inst['InstanceId']} ({name}) - {inst['InstanceType']} - EBS: {ebs_size}GB (~${ebs_size * 0.08:.2f}/mc)")

if __name__ == '__main__':
    find_unused_ebs_volumes()
    find_unused_elastic_ips()
    find_old_snapshots()
    find_stopped_instances()

Skrypt 5: Bulk tagging zasobów

import boto3

ec2 = boto3.resource('ec2', region_name='eu-central-1')

# Dodaj brakujące tagi do wszystkich instancji bez tagu Environment
for instance in ec2.instances.all():
    tags = {t['Key']: t['Value'] for t in (instance.tags or [])}

    if 'Environment' not in tags:
        # Zgadnij environment po nazwie
        name = tags.get('Name', '').lower()
        if 'prod' in name:
            env = 'production'
        elif 'stag' in name:
            env = 'staging'
        else:
            env = 'development'

        instance.create_tags(Tags=[{'Key': 'Environment', 'Value': env}])
        print(f"Tagged {instance.id} ({tags.get('Name', '?')}) → Environment={env}")

Skrypt 6: S3 bucket analyzer

import boto3
from datetime import datetime, timezone, timedelta

s3 = boto3.client('s3', region_name='eu-central-1')
cw = boto3.client('cloudwatch', region_name='eu-central-1')

def analyze_buckets():
    buckets = s3.list_buckets()['Buckets']
    print(f"=== ANALIZA S3 BUCKETÓW ({len(buckets)}) ===\n")

    for bucket in buckets:
        name = bucket['Name']

        # Rozmiar z CloudWatch
        try:
            size_response = cw.get_metric_statistics(
                Namespace='AWS/S3',
                MetricName='BucketSizeBytes',
                Dimensions=[
                    {'Name': 'BucketName', 'Value': name},
                    {'Name': 'StorageType', 'Value': 'StandardStorage'}
                ],
                StartTime=datetime.now(timezone.utc) - timedelta(days=2),
                EndTime=datetime.now(timezone.utc),
                Period=86400,
                Statistics=['Average']
            )
            size_gb = size_response['Datapoints'][0]['Average'] / (1024**3) if size_response['Datapoints'] else 0
        except Exception:
            size_gb = 0

        # Sprawdź publiczny dostęp
        try:
            acl = s3.get_bucket_acl(Bucket=name)
            public = any(
                g.get('URI', '') == 'http://acs.amazonaws.com/groups/global/AllUsers'
                for g in acl['Grants']
            )
        except Exception:
            public = False

        # Sprawdź versioning
        try:
            versioning = s3.get_bucket_versioning(Bucket=name)
            ver_status = versioning.get('Status', 'Disabled')
        except Exception:
            ver_status = 'Unknown'

        # Sprawdź lifecycle
        try:
            s3.get_bucket_lifecycle_configuration(Bucket=name)
            lifecycle = 'Tak'
        except Exception:
            lifecycle = 'Nie'

        cost = size_gb * 0.023  # S3 Standard pricing
        status = "PUBLICZNY!" if public else "OK"

        print(f"  {name}")
        print(f"    Rozmiar: {size_gb:.2f} GB (~${cost:.2f}/mc)")
        print(f"    Versioning: {ver_status} | Lifecycle: {lifecycle} | {status}")
        print()

analyze_buckets()
Dwa ograniczenia tego skryptu: sprawdzanie ACL nie wykryje bucketów upublicznionych przez bucket policy (od kwietnia 2023 ACL-e są w nowych bucketach domyślnie wyłączone) - pełny obraz dają get_bucket_policy_status() i get_public_access_block(). Po drugie, metryka BucketSizeBytes jest per region - dla bucketów spoza eu-central-1 skrypt pokaże 0 GB.

Więcej o S3 i presigned URLs w praktycznym poradniku S3.

Skrypt 7: Security Group audit

import boto3

ec2 = boto3.client('ec2', region_name='eu-central-1')

def audit_security_groups():
    """Znajdź niebezpieczne reguły Security Groups"""
    sgs = ec2.describe_security_groups()['SecurityGroups']
    issues = []

    for sg in sgs:
        for rule in sg['IpPermissions']:
            for ip_range in rule.get('IpRanges', []):
                cidr = ip_range.get('CidrIp', '')
                from_port = rule.get('FromPort', 0)
                to_port = rule.get('ToPort', 65535)

                # SSH otwarty na świat
                if cidr == '0.0.0.0/0' and from_port <= 22 <= to_port:
                    issues.append({
                        'sg': sg['GroupId'],
                        'name': sg['GroupName'],
                        'issue': f'SSH (22) otwarty na 0.0.0.0/0',
                        'severity': 'CRITICAL'
                    })

                # RDP otwarty na świat
                if cidr == '0.0.0.0/0' and from_port <= 3389 <= to_port:
                    issues.append({
                        'sg': sg['GroupId'],
                        'name': sg['GroupName'],
                        'issue': f'RDP (3389) otwarty na 0.0.0.0/0',
                        'severity': 'CRITICAL'
                    })

                # Baza danych otwarta na świat
                db_ports = [3306, 5432, 1433, 27017, 6379]
                for port in db_ports:
                    if cidr == '0.0.0.0/0' and from_port <= port <= to_port:
                        issues.append({
                            'sg': sg['GroupId'],
                            'name': sg['GroupName'],
                            'issue': f'Port bazy danych ({port}) otwarty na 0.0.0.0/0',
                            'severity': 'HIGH'
                        })

    print(f"=== SECURITY GROUP AUDIT ({len(issues)} problemów) ===\n")
    for issue in sorted(issues, key=lambda x: x['severity']):
        print(f"  [{issue['severity']}] {issue['sg']} ({issue['name']})")
        print(f"    {issue['issue']}\n")

    if not issues:
        print("  Brak problemów - Security Groups wyglądają dobrze!")

audit_security_groups()

Skrypt 8: Lambda - auto-stop instancji dev wieczorem

Oszczędzaj pieniądze na instancjach dev/test, które nie muszą działać 24/7:

import boto3

ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    """Lambda triggered by EventBridge cron: cron(0 18 ? * MON-FRI *)"""
    # Znajdź instancje z tagiem AutoStop=true
    response = ec2.describe_instances(
        Filters=[
            {'Name': 'tag:AutoStop', 'Values': ['true']},
            {'Name': 'instance-state-name', 'Values': ['running']}
        ]
    )

    instance_ids = []
    for res in response['Reservations']:
        for inst in res['Instances']:
            instance_ids.append(inst['InstanceId'])
            tags = {t['Key']: t['Value'] for t in inst.get('Tags', [])}
            print(f"Stopping: {inst['InstanceId']} ({tags.get('Name', '?')}) - {inst['InstanceType']}")

    if instance_ids:
        ec2.stop_instances(InstanceIds=instance_ids)
        print(f"Zatrzymano {len(instance_ids)} instancji")
    else:
        print("Brak instancji do zatrzymania")

    return {'stopped': len(instance_ids)}
Oszczędność: Instancja t3.medium działająca 24/7 kosztuje ~$30/mc. Działająca 10h/dzień (pn-pt) kosztuje ~$9/mc. Auto-stop + auto-start = 70% oszczędności na środowiskach dev/test.

Uruchamianie skryptów jako Lambda + EventBridge

Każdy z powyższych skryptów możesz uruchomić jako funkcję Lambda na harmonogramie:

Pamiętaj o dwóch rzeczach: execution role Lambdy musi mieć uprawnienia do wywoływanych API (np. ec2:DescribeInstances, ec2:CreateSnapshot, ce:GetCostAndUsage), a domyślny timeout Lambdy to 3 sekundy - dla tych skryptów ustaw 30-60 sekund.
# Utwórz EventBridge rule (codziennie o 7:00 UTC)
aws events put-rule \
  --name "daily-aws-audit" \
  --schedule-expression "cron(0 7 * * ? *)" \
  --state ENABLED

# Podepnij do Lambda
aws events put-targets \
  --rule "daily-aws-audit" \
  --targets "Id"="1","Arn"="arn:aws:lambda:eu-central-1:123456789:function:aws-audit"

# Daj EventBridge pozwolenie na invoke Lambda
aws lambda add-permission \
  --function-name aws-audit \
  --statement-id eventbridge-invoke \
  --action lambda:InvokeFunction \
  --principal events.amazonaws.com \
  --source-arn arn:aws:events:eu-central-1:123456789:rule/daily-aws-audit

FAQ - najczęstsze pytania o Boto3

Boto3 client vs resource - kiedy co używać?

Client to niskopoziomowe API (1:1 z AWS API), zwraca słowniki. Resource to wysokopoziomowe API, zwraca obiekty z metodami. Używaj resource gdy jest dostępny (EC2, S3, DynamoDB) - prostszy kod. Używaj client gdy resource nie obsługuje danej operacji lub potrzebujesz pełnej kontroli. Nie wszystkie serwisy mają resource (np. Cost Explorer ma tylko client).

Jak obsłużyć paginację w Boto3?

Wiele operacji AWS zwraca wyniki stronicowane. Zamiast ręcznie obsługiwać NextToken, użyj paginatora: paginator = client.get_paginator('describe_instances'), potem for page in paginator.paginate(): .... To automatycznie pobiera wszystkie strony wyników.

Czy mogę użyć Boto3 w Lambda?

Tak! Lambda ma Boto3 preinstalowane - nie musisz dodawać go do requirements.txt. Pamiętaj tylko, że wersja w Lambda może być starsza. Jeśli potrzebujesz najnowszej - dodaj boto3 jako Lambda Layer lub do paczki deployment.

Jak testować skrypty Boto3 bez ryzyka na produkcji?

Trzy podejścia: 1) Osobne konto AWS (dev) - najlepsze. 2) Biblioteka moto (mock AWS services w testach). 3) Dry-run mode - wiele operacji EC2 obsługuje parametr DryRun=True, który waliduje permissions bez wykonywania akcji.

Co dalej?

  1. Zainstaluj Boto3 i uruchom skrypt tag audit na swoim koncie.
  2. Skonfiguruj auto-snapshoty jako Lambda + EventBridge.
  3. Ustaw cotygodniowy raport kosztów na email.
  4. Dodaj auto-stop dla instancji dev/test.
  5. Rozbudowuj skrypty o własne potrzeby - Boto3 daje dostęp do KAŻDEGO serwisu AWS.

Automatyzacja to supermocy Cloud Engineera. Zamiast klikać w konsolę, piszesz skrypt raz i działa na zawsze. Jeśli chcesz automatyzować na poziomie infrastruktury (nie skryptów), sprawdź Terraform + AWS. A jeśli planujesz karierę w chmurze, przeczytaj jak zostać Cloud Engineerem.

Emil Kowalczyk

Pasjonat chmury i twórca CloudManiak.pl. Na co dzień MSP Engineer w amerykańskiej firmie ClearScale (AWS Premier Tier Partner). Pomagam osobom wchodzącym do świata chmury zdobywać wiedzę i certyfikaty.