cal/paper-dynasty-database
1
0
Fork 0
Code Issues 8 Pull Requests 3 Actions Packages Projects Releases Wiki Activity

feat: add card_storage S3 upload utility for variant cards

Browse Source 
New service with S3 upload functions for the refractor card art
pipeline. backfill_variant_image_url reads rendered PNGs from disk,
uploads to S3, and sets image_url on BattingCard/PitchingCard rows.
18 tests covering key construction, URL formatting, upload params,
and error swallowing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Cal Corum 2026-04-06 17:05:53 -05:00 committed by cal
parent 088d30b96b
commit 534d50f1a8
3 changed files with 552 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

201
app/services/card_storage.py Normal file
View File

@ -0,0 +1,201 @@
"""
card_storage.py — S3 upload utility for variant card images.
Public API
----------
get_s3_client()
Create and return a boto3 S3 client using ambient AWS credentials
(environment variables or instance profile).
build_s3_key(cardset_id, player_id, variant, card_type)
Construct the S3 object key for a variant card image.
build_s3_url(s3_key, render_date)
Return the full HTTPS S3 URL with a cache-busting date query param.
upload_card_to_s3(s3_client, png_bytes, s3_key)
Upload raw PNG bytes to S3 with correct ContentType and CacheControl headers.
backfill_variant_image_url(player_id, variant, card_type, cardset_id, png_path)
End-to-end: read PNG from disk, upload to S3, update BattingCard or
PitchingCard.image_url in the database. All exceptions are caught and
logged; this function never raises (safe to call as a background task).
Design notes
------------
- backfill_variant_image_url uses a lazy import for BattingCard/PitchingCard
to avoid circular imports at module load time (db_engine imports routers
indirectly via the app module in some paths).
- S3 credentials are resolved from the environment by boto3 at call time;
no credentials are hard-coded here.
- The cache-bust ?d= param matches the card-creation pipeline convention so
that clients can compare URLs across pipelines.
"""
import logging
from datetime import date
import boto3
logger = logging.getLogger(__name__)
S3_BUCKET = "paper-dynasty"
S3_REGION = "us-east-1"
def get_s3_client():
"""Create and return a boto3 S3 client for the configured region.
Credentials are resolved by boto3 from the standard chain:
environment variables → ~/.aws/credentials → instance profile.
Returns:
A boto3 S3 client instance.
"""
return boto3.client("s3", region_name=S3_REGION)
def build_s3_key(cardset_id: int, player_id: int, variant: int, card_type: str) -> str:
"""Construct the S3 object key for a variant card image.
Key format:
cards/cardset-{csid:03d}/player-{pid}/v{variant}/{card_type}card.png
Args:
cardset_id: Numeric cardset ID (zero-padded to 3 digits).
player_id: Player ID.
variant: Variant number (0 = base, 1-4 = refractor tiers).
card_type: Either "batting" or "pitching".
Returns:
The S3 object key string.
"""
return (
f"cards/cardset-{cardset_id:03d}/player-{player_id}"
f"/v{variant}/{card_type}card.png"
)
def build_s3_url(s3_key: str, render_date: date) -> str:
"""Return the full HTTPS S3 URL for a card image with a cache-bust param.
URL format:
https://{bucket}.s3.{region}.amazonaws.com/{key}?d={date}
The ?d= query param matches the card-creation pipeline convention so that
clients invalidate their cache after each re-render.
Args:
s3_key: S3 object key (from build_s3_key).
render_date: The date the card was rendered, used for cache-busting.
Returns:
Full HTTPS URL string.
"""
base_url = f"https://{S3_BUCKET}.s3.{S3_REGION}.amazonaws.com"
date_str = render_date.strftime("%Y-%m-%d")
return f"{base_url}/{s3_key}?d={date_str}"
def upload_card_to_s3(s3_client, png_bytes: bytes, s3_key: str) -> None:
"""Upload raw PNG bytes to S3 with the standard card image headers.
Sets ContentType=image/png and CacheControl=public, max-age=300 (5 min)
so that CDN and browser caches are refreshed within a short window after
a re-render.
Args:
s3_client: A boto3 S3 client (from get_s3_client).
png_bytes: Raw PNG image bytes.
s3_key: S3 object key (from build_s3_key).
Returns:
None
"""
s3_client.put_object(
Bucket=S3_BUCKET,
Key=s3_key,
Body=png_bytes,
ContentType="image/png",
CacheControl="public, max-age=300",
)
def backfill_variant_image_url(
player_id: int,
variant: int,
card_type: str,
cardset_id: int,
png_path: str,
) -> None:
"""Read a rendered PNG from disk, upload it to S3, and update the DB row.
Determines the correct card model (BattingCard or PitchingCard) from
card_type, then:
1. Reads PNG bytes from png_path.
2. Uploads to S3 via upload_card_to_s3.
3. Fetches the card row by (player_id, variant).
4. Sets image_url to the new S3 URL and calls save().
All exceptions are caught and logged — this function is intended to be
called as a background task and must never propagate exceptions.
Args:
player_id: Player ID used to locate the card row.
variant: Variant number (matches the card row's variant field).
card_type: "batting" or "pitching" — selects the model.
cardset_id: Cardset ID used for the S3 key.
png_path: Absolute path to the rendered PNG file on disk.
Returns:
None
"""
# Lazy import to avoid circular imports at module load time
from app.db_engine import BattingCard, PitchingCard
try:
# 1. Read PNG from disk
with open(png_path, "rb") as f:
png_bytes = f.read()
# 2. Build key and upload
s3_key = build_s3_key(
cardset_id=cardset_id,
player_id=player_id,
variant=variant,
card_type=card_type,
)
s3_client = get_s3_client()
upload_card_to_s3(s3_client, png_bytes, s3_key)
# 3. Build URL with today's date for cache-busting
image_url = build_s3_url(s3_key, render_date=date.today())
# 4. Locate the card row and update image_url
if card_type == "batting":
card = BattingCard.get(
BattingCard.player == player_id, BattingCard.variant == variant
)
else:
card = PitchingCard.get(
PitchingCard.player == player_id, PitchingCard.variant == variant
)
card.image_url = image_url
card.save()
logger.info(
"backfill_variant_image_url: updated %s card player=%s variant=%s url=%s",
card_type,
player_id,
variant,
image_url,
)
except Exception:
logger.exception(
"backfill_variant_image_url: failed for player=%s variant=%s card_type=%s",
player_id,
variant,
card_type,
)

1
requirements.txt
View File

@ -12,3 +12,4 @@ requests==2.32.3
html2image==2.0.6
jinja2==3.1.4
playwright==1.45.1
boto3>=1.34.0

350
tests/test_card_storage.py Normal file
View File

@ -0,0 +1,350 @@
"""
Unit tests for app/services/card_storage.py — S3 upload utility.
This module covers:
- S3 key construction for variant cards (batting, pitching, zero-padded cardset)
- Full S3 URL construction with cache-busting date param
- put_object call validation (correct params, return value)
- End-to-end backfill: read PNG from disk, upload to S3, update DB row
Why we test S3 key construction separately:
The key format is a contract used by both the renderer and the URL builder.
Validating it in isolation catches regressions before they corrupt stored URLs.
Why we test URL construction separately:
The cache-bust param (?d=...) must be appended consistently so that clients
invalidate cached images after a re-render. Testing it independently prevents
the formatter from silently changing.
Why we test upload params:
ContentType and CacheControl must be set exactly so that S3 serves images
with the correct headers. A missing header is a silent misconfiguration.
Why we test backfill error swallowing:
The backfill function is called as a background task — it must never raise
exceptions that would abort a card render response. We verify that S3 failures
and missing files are both silently logged, not propagated.
Test isolation:
All tests use unittest.mock; no real S3 calls or DB connections are made.
The `backfill_variant_image_url` tests patch `get_s3_client` and the DB
model classes at the card_storage module level so lazy imports work correctly.
"""
import os
from datetime import date
from unittest.mock import MagicMock, patch
# Set env before importing module so db_engine doesn't try to connect
os.environ.setdefault("DATABASE_TYPE", "postgresql")
os.environ.setdefault("POSTGRES_PASSWORD", "test-dummy")
from app.services.card_storage import (
build_s3_key,
build_s3_url,
upload_card_to_s3,
backfill_variant_image_url,
S3_BUCKET,
S3_REGION,
)
# ---------------------------------------------------------------------------
# TestBuildS3Key
# ---------------------------------------------------------------------------
class TestBuildS3Key:
"""Tests for build_s3_key — S3 object key construction.
The key format must match the existing card-creation pipeline so that
the database API and card-creation tool write to the same S3 paths.
"""
def test_batting_card_key(self):
"""batting card type produces 'battingcard.png' in the key."""
key = build_s3_key(cardset_id=27, player_id=42, variant=1, card_type="batting")
assert key == "cards/cardset-027/player-42/v1/battingcard.png"
def test_pitching_card_key(self):
"""pitching card type produces 'pitchingcard.png' in the key."""
key = build_s3_key(cardset_id=27, player_id=99, variant=2, card_type="pitching")
assert key == "cards/cardset-027/player-99/v2/pitchingcard.png"
def test_cardset_zero_padded_to_three_digits(self):
"""Single-digit cardset IDs are zero-padded to three characters."""
key = build_s3_key(cardset_id=5, player_id=1, variant=0, card_type="batting")
assert "cardset-005" in key
def test_cardset_two_digit_zero_padded(self):
"""Two-digit cardset IDs are zero-padded correctly."""
key = build_s3_key(cardset_id=27, player_id=1, variant=0, card_type="batting")
assert "cardset-027" in key
def test_cardset_three_digit_no_padding(self):
"""Three-digit cardset IDs are not altered."""
key = build_s3_key(cardset_id=100, player_id=1, variant=0, card_type="batting")
assert "cardset-100" in key
def test_variant_included_in_key(self):
"""Variant number is included in the path so variants have distinct keys."""
key_v0 = build_s3_key(
cardset_id=27, player_id=1, variant=0, card_type="batting"
)
key_v3 = build_s3_key(
cardset_id=27, player_id=1, variant=3, card_type="batting"
)
assert "/v0/" in key_v0
assert "/v3/" in key_v3
assert key_v0 != key_v3
# ---------------------------------------------------------------------------
# TestBuildS3Url
# ---------------------------------------------------------------------------
class TestBuildS3Url:
"""Tests for build_s3_url — full URL construction with cache-bust param.
The URL format must be predictable so clients can construct and verify
image URLs without querying the database.
"""
def test_url_contains_bucket_and_region(self):
"""URL includes bucket name and region in the S3 hostname."""
key = "cards/cardset-027/player-42/v1/battingcard.png"
render_date = date(2026, 4, 6)
url = build_s3_url(key, render_date)
assert S3_BUCKET in url
assert S3_REGION in url
def test_url_contains_s3_key(self):
"""URL path includes the full S3 key."""
key = "cards/cardset-027/player-42/v1/battingcard.png"
render_date = date(2026, 4, 6)
url = build_s3_url(key, render_date)
assert key in url
def test_url_has_cache_bust_param(self):
"""URL ends with ?d=<render_date> for cache invalidation."""
key = "cards/cardset-027/player-42/v1/battingcard.png"
render_date = date(2026, 4, 6)
url = build_s3_url(key, render_date)
assert "?d=2026-04-06" in url
def test_url_format_full(self):
"""Full URL matches expected S3 pattern exactly."""
key = "cards/cardset-027/player-1/v0/battingcard.png"
render_date = date(2025, 11, 8)
url = build_s3_url(key, render_date)
expected = (
f"https://{S3_BUCKET}.s3.{S3_REGION}.amazonaws.com/{key}?d=2025-11-08"
)
assert url == expected
# ---------------------------------------------------------------------------
# TestUploadCardToS3
# ---------------------------------------------------------------------------
class TestUploadCardToS3:
"""Tests for upload_card_to_s3 — S3 put_object call validation.
We verify the exact parameters passed to put_object so that S3 serves
images with the correct Content-Type and Cache-Control headers.
"""
def test_put_object_called_with_correct_params(self):
"""put_object is called once with bucket, key, body, ContentType, CacheControl."""
mock_client = MagicMock()
png_bytes = b"\x89PNG\r\n\x1a\n"
s3_key = "cards/cardset-027/player-42/v1/battingcard.png"
upload_card_to_s3(mock_client, png_bytes, s3_key)
mock_client.put_object.assert_called_once_with(
Bucket=S3_BUCKET,
Key=s3_key,
Body=png_bytes,
ContentType="image/png",
CacheControl="public, max-age=300",
)
def test_upload_returns_none(self):
"""upload_card_to_s3 returns None (callers should not rely on a return value)."""
mock_client = MagicMock()
result = upload_card_to_s3(mock_client, b"PNG", "some/key.png")
assert result is None
# ---------------------------------------------------------------------------
# TestBackfillVariantImageUrl
# ---------------------------------------------------------------------------
class TestBackfillVariantImageUrl:
"""Tests for backfill_variant_image_url — end-to-end disk→S3→DB path.
The function is fire-and-forget: it reads a PNG from disk, uploads to S3,
then updates the appropriate card model's image_url. All errors are caught
and logged; the function must never raise.
Test strategy:
- Use tmp_path for temporary PNG files so no filesystem state leaks.
- Patch get_s3_client at the module level to intercept the S3 call.
- Patch BattingCard/PitchingCard at the module level (lazy import target).
"""
def test_batting_card_image_url_updated(self, tmp_path):
"""BattingCard.image_url is updated after a successful upload."""
png_path = tmp_path / "card.png"
png_path.write_bytes(b"\x89PNG\r\n\x1a\n fake png data")
mock_s3 = MagicMock()
mock_card = MagicMock()
with (
patch("app.services.card_storage.get_s3_client", return_value=mock_s3),
patch("app.db_engine.BattingCard") as MockBatting,
):
MockBatting.get.return_value = mock_card
backfill_variant_image_url(
player_id=42,
variant=1,
card_type="batting",
cardset_id=27,
png_path=str(png_path),
)
MockBatting.get.assert_called_once_with(
MockBatting.player == 42, MockBatting.variant == 1
)
assert mock_card.image_url is not None
mock_card.save.assert_called_once()
def test_pitching_card_image_url_updated(self, tmp_path):
"""PitchingCard.image_url is updated after a successful upload."""
png_path = tmp_path / "card.png"
png_path.write_bytes(b"\x89PNG\r\n\x1a\n fake png data")
mock_s3 = MagicMock()
mock_card = MagicMock()
with (
patch("app.services.card_storage.get_s3_client", return_value=mock_s3),
patch("app.db_engine.PitchingCard") as MockPitching,
):
MockPitching.get.return_value = mock_card
backfill_variant_image_url(
player_id=99,
variant=2,
card_type="pitching",
cardset_id=27,
png_path=str(png_path),
)
MockPitching.get.assert_called_once_with(
MockPitching.player == 99, MockPitching.variant == 2
)
assert mock_card.image_url is not None
mock_card.save.assert_called_once()
def test_s3_upload_called_with_png_bytes(self, tmp_path):
"""The PNG bytes read from disk are passed to put_object."""
png_bytes = b"\x89PNG\r\n\x1a\n real png content"
png_path = tmp_path / "card.png"
png_path.write_bytes(png_bytes)
mock_s3 = MagicMock()
with (
patch("app.services.card_storage.get_s3_client", return_value=mock_s3),
patch("app.db_engine.BattingCard") as MockBatting,
):
MockBatting.get.return_value = MagicMock()
backfill_variant_image_url(
player_id=1,
variant=0,
card_type="batting",
cardset_id=5,
png_path=str(png_path),
)
mock_s3.put_object.assert_called_once()
call_kwargs = mock_s3.put_object.call_args.kwargs
assert call_kwargs["Body"] == png_bytes
def test_s3_error_is_swallowed(self, tmp_path):
"""If S3 raises an exception, backfill swallows it and returns normally.
The function is called as a background task — it must never propagate
exceptions that would abort the calling request handler.
"""
png_path = tmp_path / "card.png"
png_path.write_bytes(b"PNG data")
mock_s3 = MagicMock()
mock_s3.put_object.side_effect = Exception("S3 connection refused")
with (
patch("app.services.card_storage.get_s3_client", return_value=mock_s3),
patch("app.db_engine.BattingCard"),
):
# Must not raise
backfill_variant_image_url(
player_id=1,
variant=0,
card_type="batting",
cardset_id=27,
png_path=str(png_path),
)
def test_missing_file_is_swallowed(self, tmp_path):
"""If the PNG file does not exist, backfill swallows the error and returns.
Render failures may leave no file on disk; the background task must
handle this gracefully rather than crashing the request.
"""
missing_path = str(tmp_path / "nonexistent.png")
with (
patch("app.services.card_storage.get_s3_client"),
patch("app.db_engine.BattingCard"),
):
# Must not raise
backfill_variant_image_url(
player_id=1,
variant=0,
card_type="batting",
cardset_id=27,
png_path=missing_path,
)
def test_db_error_is_swallowed(self, tmp_path):
"""If the DB save raises, backfill swallows it and returns normally."""
png_path = tmp_path / "card.png"
png_path.write_bytes(b"PNG data")
mock_s3 = MagicMock()
mock_card = MagicMock()
mock_card.save.side_effect = Exception("DB connection lost")
with (
patch("app.services.card_storage.get_s3_client", return_value=mock_s3),
patch("app.db_engine.BattingCard") as MockBatting,
):
MockBatting.get.return_value = mock_card
# Must not raise
backfill_variant_image_url(
player_id=1,
variant=0,
card_type="batting",
cardset_id=27,
png_path=str(png_path),
)