AVAX S3 — Python Developer Guide

Praktický průvodce pro přístup k S3 úložišti z Python aplikací distribuovaných přes platformu AVAX.

---

Rychlý start

pip install boto3 requests

import boto3
import requests

# 1. Získej dočasné S3 credentials přes AVAX API
token = get_storage_token(access_token="...", scope="private")

# 2. Připoj se na S3
s3 = boto3.client(
    "s3",
    endpoint_url=token["endpoint"],
    aws_access_key_id=token["access_key"],
    aws_secret_access_key=token["secret_key"],
    aws_session_token=token["session_token"],
)

# 3. Nahraj soubor
s3.upload_file("local_file.db", token["bucket"], f"{token['prefix']}data/evidence.db")

# 4. Stáhni soubor
s3.download_file(token["bucket"], f"{token['prefix']}data/evidence.db", "local_copy.db")

---

Autentizace — získání S3 tokenu

Aplikace nikdy nepracuje s permanentními S3 klíči. Vždy žádá backend o krátkodobý token (TTL 1 hodina).

# auth.py

import requests
from dataclasses import dataclass
from datetime import datetime

API_BASE = "https://api.avaxis.cz"  # dev: "http://192.168.1.55:8000"

@dataclass
class StorageToken:
    endpoint:      str
    bucket:        str
    prefix:        str           # základní prefix pro tento scope
    access_key:    str
    secret_key:    str
    session_token: str
    expires_at:    datetime
    can_write:     bool

def get_storage_token(access_token: str, scope: str, **kwargs) -> StorageToken:
    """
    scope:
      'private'          → soukromý adresář uživatele
      'shared'           → sdílený adresář (vyžaduje dir_id=<uuid>)
      'app'              → data aplikace  (vyžaduje app_slug=<str>)
      'app_backup'       → zálohy (read-only, vyžaduje app_slug=<str>)
    """
    params = {"scope": scope, **kwargs}
    resp = requests.get(
        f"{API_BASE}/storage/token",
        headers={"Authorization": f"Bearer {access_token}"},
        params=params,
        timeout=10,
    )
    resp.raise_for_status()
    data = resp.json()
    return StorageToken(
        endpoint=data["endpoint"],
        bucket=data["bucket"],
        prefix=data["prefix"],
        access_key=data["access_key"],
        secret_key=data["secret_key"],
        session_token=data["session_token"],
        expires_at=datetime.fromisoformat(data["expires_at"]),
        can_write=data.get("can_write", True),
    )

---

S3Client — obalová třída

Doporučujeme používat tuto třídu která řeší: - automatické obnovení tokenu před expirací - správné prefixování cest - kontext managera

# s3client.py

import boto3
import requests
from datetime import datetime, timedelta, timezone
from pathlib import Path

class AvaxS3Client:
    """
    Klient pro přístup k AVAX S3 úložišti.
    Automaticky obnovuje dočasné credentials 5 minut před expirací.

    Použití:
        with AvaxS3Client(access_token, scope="private") as s3:
            s3.upload("local.db", "data/evidence.db")
            s3.download("data/evidence.db", "local.db")
            files = s3.list("data/")
    """

    RENEW_BEFORE = timedelta(minutes=5)

    def __init__(self, access_token: str, scope: str, **scope_kwargs):
        self._access_token = access_token
        self._scope        = scope
        self._scope_kwargs = scope_kwargs
        self._token        = None
        self._client       = None

    # ── Context manager ──────────────────────────────────────────────────────

    def __enter__(self):
        self._refresh()
        return self

    def __exit__(self, *_):
        self._client = None

    # ── Veřejné metody ───────────────────────────────────────────────────────

    def upload(self, local_path: str | Path, remote_path: str,
               extra_args: dict | None = None) -> None:
        """Nahraje soubor. remote_path je relativní k prefixu tokenu."""
        self._ensure_fresh()
        key = self._key(remote_path)
        self._client.upload_file(str(local_path), self._token.bucket, key,
                                 ExtraArgs=extra_args or {})

    def upload_bytes(self, data: bytes, remote_path: str) -> None:
        """Nahraje bytes přímo do S3 bez dočasného souboru."""
        self._ensure_fresh()
        self._client.put_object(Bucket=self._token.bucket,
                                Key=self._key(remote_path), Body=data)

    def download(self, remote_path: str, local_path: str | Path) -> None:
        """Stáhne soubor z S3."""
        self._ensure_fresh()
        Path(local_path).parent.mkdir(parents=True, exist_ok=True)
        self._client.download_file(self._token.bucket, self._key(remote_path),
                                   str(local_path))

    def download_bytes(self, remote_path: str) -> bytes:
        """Vrátí obsah souboru jako bytes."""
        self._ensure_fresh()
        obj = self._client.get_object(Bucket=self._token.bucket,
                                      Key=self._key(remote_path))
        return obj["Body"].read()

    def list(self, prefix: str = "") -> list[dict]:
        """
        Vypíše objekty pod daným prefixem.
        Vrátí: [{ key, size, last_modified }]
        """
        self._ensure_fresh()
        full_prefix = self._key(prefix)
        paginator = self._client.get_paginator("list_objects_v2")
        result = []
        for page in paginator.paginate(Bucket=self._token.bucket, Prefix=full_prefix):
            for obj in page.get("Contents", []):
                result.append({
                    "key":           obj["Key"].removeprefix(self._token.prefix),
                    "size":          obj["Size"],
                    "last_modified": obj["LastModified"],
                })
        return result

    def exists(self, remote_path: str) -> bool:
        """Ověří zda soubor existuje."""
        self._ensure_fresh()
        try:
            self._client.head_object(Bucket=self._token.bucket,
                                     Key=self._key(remote_path))
            return True
        except self._client.exceptions.ClientError:
            return False

    def presigned_url(self, remote_path: str, ttl: int = 900) -> str:
        """Vygeneruje presigned GET URL platný `ttl` sekund (max 900)."""
        self._ensure_fresh()
        return self._client.generate_presigned_url(
            "get_object",
            Params={"Bucket": self._token.bucket, "Key": self._key(remote_path)},
            ExpiresIn=ttl,
        )

    # ── Interní ──────────────────────────────────────────────────────────────

    def _key(self, path: str) -> str:
        """Sestaví plný S3 klíč: prefix + path (bez dvojitých lomítek)."""
        return self._token.prefix.rstrip("/") + "/" + path.lstrip("/")

    def _ensure_fresh(self):
        expires = self._token.expires_at.replace(tzinfo=timezone.utc) \
                  if self._token.expires_at.tzinfo is None \
                  else self._token.expires_at
        if datetime.now(timezone.utc) >= expires - self.RENEW_BEFORE:
            self._refresh()

    def _refresh(self):
        self._token = get_storage_token(
            self._access_token, self._scope, **self._scope_kwargs
        )
        self._client = boto3.client(
            "s3",
            endpoint_url=self._token.endpoint,
            aws_access_key_id=self._token.access_key,
            aws_secret_access_key=self._token.secret_key,
            aws_session_token=self._token.session_token,
        )

---

Příklady použití

Soukromý adresář uživatele

with AvaxS3Client(access_token, scope="private") as s3:
    # Uložení lokální SQLite databáze
    s3.upload("data/evidence.db", "evidence.db")

    # Načtení zpět
    s3.download("evidence.db", "data/evidence_restored.db")

    # Výpis souborů
    for f in s3.list(""):
        print(f["key"], f["size"])

Sdílený adresář (admin musí udělit přístup)

# Nejdřív zjisti dostupné sdílené adresáře
dirs = requests.get(
    f"{API_BASE}/storage/directories",
    headers={"Authorization": f"Bearer {access_token}"},
).json()
# [{ "id": "...", "name": "Účetní dokumenty", "can_write": True }, ...]

dir_id = dirs[0]["id"]

with AvaxS3Client(access_token, scope="shared", dir_id=dir_id) as s3:
    # Nahrání sdíleného dokumentu
    s3.upload("faktury/2025-01.pdf", "faktury/2025-01.pdf")

    # Čtení dokumentu jiného uživatele ve stejném adresáři
    data = s3.download_bytes("faktury/2025-02.pdf")

Data aplikace (AVAX SDK)

# Při použití AVAX SDK je token automatický:
from avaxis_sdk import AvaxApp

app = AvaxApp(slug="moje-ucetnictvi")
app.connect()

# SDK poskytuje AvaxS3Client přímo
with app.storage.app_data() as s3:
    s3.upload(app.paths.data / "db.sqlite", "db.sqlite")

# Nebo ručně:
with AvaxS3Client(app.access_token, scope="app", app_slug="moje-ucetnictvi") as s3:
    s3.upload("db.sqlite", "db.sqlite")

Upload bytes (bez dočasného souboru)

import json

with AvaxS3Client(access_token, scope="private") as s3:
    # Uložení konfigurace jako JSON
    config = {"theme": "dark", "lang": "cs", "version": "2.1"}
    s3.upload_bytes(json.dumps(config).encode(), "settings/config.json")

    # Načtení
    raw = s3.download_bytes("settings/config.json")
    config = json.loads(raw)

Presigned URL pro stažení v prohlížeči

with AvaxS3Client(access_token, scope="shared", dir_id=dir_id) as s3:
    url = s3.presigned_url("reports/report-2025-Q1.pdf", ttl=600)
    # url je platné 10 minut — pošli ho uživateli pro stažení
    print(url)

---

Správa sdílených adresářů (admin)

import requests

headers = {"Authorization": f"Bearer {admin_token}", "Content-Type": "application/json"}

# Vytvoření adresáře
resp = requests.post(f"{API_BASE}/storage/directories", headers=headers, json={
    "name": "Účetní dokumenty 2025",
    "description": "Faktury, výpisy, přiznání",
})
dir_id = resp.json()["id"]

# Udělení přístupu konkrétnímu uživateli (read+write)
requests.post(f"{API_BASE}/storage/directories/{dir_id}/access", headers=headers, json={
    "user_id":   "uuid-uzivatele",
    "can_read":  True,
    "can_write": True,
})

# Udělení přístupu celé roli (read-only)
requests.post(f"{API_BASE}/storage/directories/{dir_id}/access", headers=headers, json={
    "role_id":   "uuid-role-ucetni",
    "can_read":  True,
    "can_write": False,
})

# Udělení přístupu všem ve firmě (read-only)
requests.post(f"{API_BASE}/storage/directories/{dir_id}/access", headers=headers, json={
    "all_company": True,
    "can_read":    True,
    "can_write":   False,
})

# Odebrání přístupu
requests.delete(f"{API_BASE}/storage/directories/{dir_id}/access/{access_id}", headers=headers)

# Výpis ACL záznamu adresáře
acl = requests.get(f"{API_BASE}/storage/directories/{dir_id}/access", headers=headers).json()

---

Scope přehled

Scope	Prefix v bucketu	TTL	Write
private	users/{user_id}/	1 h	ano
shared + dir_id	shared/custom/{dir_id}/	1 h	dle ACL
app + app_slug	apps/{slug}/data/	15 min	ano
app_backup + app_slug	apps/{slug}/backups/	15 min	ne

---

Dev server

API_BASE = "http://192.168.1.55:8000"

# Testovací přihlášení
resp = requests.post(f"{API_BASE}/auth/login", json={
    "email":    "admin@test.cz",
    "password": "Heslo123!",
})
access_token = resp.json()["access_token"]

# Test soukromého S3
with AvaxS3Client(access_token, scope="private") as s3:
    s3.upload_bytes(b"hello world", "test.txt")
    print(s3.download_bytes("test.txt"))  # b"hello world"
    print(s3.list(""))

---

Chybové stavy

import boto3.exceptions
import botocore.exceptions

try:
    with AvaxS3Client(access_token, scope="shared", dir_id=dir_id) as s3:
        s3.upload("file.pdf", "docs/file.pdf")

except requests.HTTPError as e:
    if e.response.status_code == 403:
        print("Nemáte přístup k tomuto adresáři")
    elif e.response.status_code == 401:
        print("Token vypršel — přihlaste se znovu")

except botocore.exceptions.ClientError as e:
    code = e.response["Error"]["Code"]
    if code == "NoSuchKey":
        print("Soubor neexistuje")
    elif code == "AccessDenied":
        print("S3 přístup odepřen — token mohl vypršet")
    else:
        raise

---

Kompletní requirements.txt

boto3>=1.34.0
botocore>=1.34.0
requests>=2.31.0

---

AVAX Platform | api.avaxis.cz | Aktualizováno: 2026-04-24