From e0b701e3d986c2363793bddd0cddca09f3140cce Mon Sep 17 00:00:00 2001 From: Sean N Date: Sat, 16 May 2026 09:17:18 +0200 Subject: [PATCH 1/2] fixed general issues and add test coverage --- SKILL.md | 26 +++- pyproject.toml | 2 +- src/nitro_img/__init__.py | 2 +- src/nitro_img/batch.py | 12 ++ src/nitro_img/config.py | 15 +- src/nitro_img/image.py | 215 ++++++++++++++++---------- src/nitro_img/integrations.py | 67 ++++---- src/nitro_img/loaders.py | 84 ++++++++-- src/nitro_img/operations/effects.py | 28 ++-- src/nitro_img/operations/metadata.py | 15 +- src/nitro_img/operations/overlay.py | 72 +++++---- src/nitro_img/operations/resize.py | 26 +++- src/nitro_img/operations/transform.py | 7 +- src/nitro_img/output/optimize.py | 34 ++-- src/nitro_img/pipeline.py | 15 +- src/nitro_img/placeholder.py | 35 ++--- src/nitro_img/utils.py | 14 +- tests/test_config.py | 167 ++++++++++++++++++++ tests/test_from_url.py | 118 ++++++++++++++ tests/test_review_fixes.py | 186 ++++++++++++++++++++++ 20 files changed, 907 insertions(+), 233 deletions(-) create mode 100644 tests/test_config.py create mode 100644 tests/test_from_url.py create mode 100644 tests/test_review_fixes.py diff --git a/SKILL.md b/SKILL.md index 10a99ec..42a8b94 100644 --- a/SKILL.md +++ b/SKILL.md @@ -429,7 +429,7 @@ results = batch.save( ### Available Chainable Methods -BatchImage supports the same chainable methods as Image: `resize`, `thumbnail`, `cover`, `contain`, `crop`, `rotate`, `flip`, `mirror`, `grayscale`, `strip_metadata`, `watermark`, `text_overlay`, `brightness`, `contrast`, `saturation`, `sharpen`, `blur`, `sepia`, `rounded_corners`, `jpeg`, `png`, `webp`, `format`. +BatchImage supports the same chainable methods as Image: `resize`, `thumbnail`, `cover`, `contain`, `crop`, `rotate`, `flip`, `mirror`, `grayscale`, `strip_metadata`, `watermark`, `text_overlay`, `brightness`, `contrast`, `saturation`, `sharpen`, `blur`, `sepia`, `rounded_corners`, `jpeg`, `png`, `webp`, `gif`, `format`. --- @@ -445,10 +445,12 @@ config.update( allow_upscale=False, # bool, default: False auto_orient=True, # bool, default: True - auto-apply EXIF rotation on load strip_metadata=False, # bool, default: False - auto-strip metadata on load - max_input_size=50_000_000, # int bytes, default: 50MB - max_output_dimensions=10_000, # int pixels, default: 10000 + max_input_size=50_000_000, # int bytes, default: 50MB - applies to uploads, files, bytes + max_pixels=178_956_970, # int pixels, default: ~179MP - decompression-bomb guard + max_output_dimensions=10_000, # int pixels, default: 10000 - cap on resize result width/height url_timeout=30.0, # float seconds, default: 30.0 - url_max_size=50_000_000, # int bytes, default: 50MB + url_max_size=50_000_000, # int bytes, default: 50MB - download aborts when exceeded + url_allowed_schemes=("http", "https"), # tuple[str, ...] - URL schemes accepted by from_url ) ``` @@ -614,8 +616,22 @@ img.save_responsive("static/hero/", [400, 800, 1200, 1600], name="hero") 5. **Format required for byte output.** `.to_bytes()`, `.to_base64()`, etc. need a format set (via `.jpeg()`, `.webp()`, etc.). `.save()` can infer from the file extension. -6. **auto_format is deferred.** `.auto_format()` doesn't pick the format immediately - it resolves when an output method is called. Images with alpha get PNG; photographic images get WebP or JPEG (whichever is smaller). +6. **auto_format is deferred.** `.auto_format()` doesn't pick the format immediately - it resolves when an output method is called. Images with alpha get PNG; photographic images get WebP or JPEG (whichever is smaller). `auto_format()` is honored by every output method, including `optimize()` and the Django/Flask/FastAPI response helpers. 7. **Presets are standalone.** `Image.preset.thumbnail("photo.jpg")` returns `bytes` directly - it doesn't return an `Image` instance. Presets load, process, and encode in one call. 8. **BatchImage records, doesn't chain Image instances.** Each operation on `BatchImage` records the method name and args. On `.save()`, it creates a fresh `Image` for each file and replays the recorded operations. + +9. **Resize caps via `max_output_dimensions`.** Any resize/cover/contain/thumbnail call that would produce an output wider or taller than `config.max_output_dimensions` raises `ImageSizeError`. Bump the config if you need larger output. + +10. **`responsive()` widths are measured against the post-pipeline image.** Operations queued before `.responsive(...)` execute first, so chained resizes/crops shrink the source the responsive set is derived from. Skip in-pipeline resizes if you want widths relative to the original. + +11. **`from_url` is strict.** Only `http` and `https` schemes are accepted by default (set `config.url_allowed_schemes` to extend). The download streams in chunks and aborts as soon as `config.url_max_size` is exceeded; `Content-Length` headers above the cap short-circuit the request. + +12. **Decompression bombs are blocked at load time.** `config.max_pixels` (default ~179MP, matching Pillow's own cap) rejects any source whose dimensions imply more than that many pixels before pixel data is decoded. + +13. **`grayscale()` preserves alpha.** RGBA inputs stay RGBA - the alpha channel survives the conversion. + +14. **`strip_metadata()` does not preserve animation frames.** Animated GIF/WebP inputs are collapsed to a single frame after stripping. + +15. **Position arguments are validated.** `watermark()` and `text_overlay()` raise `ValueError` for unknown `position` values. `"tiled"` is only accepted by `watermark()`. diff --git a/pyproject.toml b/pyproject.toml index 3d47cc0..f465f13 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "nitro-image" -version = "1.0.2" +version = "1.1.1" description = "Fast, friendly image processing for web apps and SaaS" readme = "README.md" license = "BSD-3-Clause" diff --git a/src/nitro_img/__init__.py b/src/nitro_img/__init__.py index e329034..9b15b8c 100644 --- a/src/nitro_img/__init__.py +++ b/src/nitro_img/__init__.py @@ -14,7 +14,7 @@ ) from .types import Format, Position, ResizeStrategy -__version__ = "1.0.2" +__version__ = "1.1.1" __all__ = [ "Image", diff --git a/src/nitro_img/batch.py b/src/nitro_img/batch.py index baf45f1..660db1f 100644 --- a/src/nitro_img/batch.py +++ b/src/nitro_img/batch.py @@ -392,6 +392,18 @@ def webp(self, quality: int | None = None) -> BatchImage: self._output_quality = quality return self + def gif(self) -> BatchImage: + """Select GIF for every image in the batch. + + Returns: + The same ``BatchImage`` for chaining. + + Example: + >>> BatchImage("frames/*.png").gif().save("out/{name}.gif") + """ + self._output_format = Format.GIF + return self + def format(self, fmt: Format | str, quality: int | None = None) -> BatchImage: """Select the output format by enum or case-insensitive string. diff --git a/src/nitro_img/config.py b/src/nitro_img/config.py index 4784f0e..def48df 100644 --- a/src/nitro_img/config.py +++ b/src/nitro_img/config.py @@ -17,8 +17,13 @@ class Config: Attributes: max_input_size: Maximum accepted input size in bytes. Larger inputs raise :class:`ImageSizeError`. + max_pixels: Maximum decoded pixel count for any loaded image. + Guards against decompression-bomb attacks where a tiny file + expands to a huge raster. Larger inputs raise + :class:`ImageSizeError`. max_output_dimensions: Maximum width or height in pixels for any - resize result. + resize result. Operations that would exceed this raise + :class:`ImageSizeError`. default_jpeg_quality: JPEG quality used when no explicit value is passed to ``.jpeg()`` or an output method. default_webp_quality: WebP quality used when no explicit value is @@ -32,7 +37,11 @@ class Config: the pipeline runs. url_timeout: Request timeout in seconds for ``Image.from_url``. url_max_size: Maximum accepted response size in bytes for - ``Image.from_url``. + ``Image.from_url``. The download is aborted as soon as this + cap is exceeded; the body is never fully buffered above it. + url_allowed_schemes: URL schemes accepted by ``Image.from_url``. + Defaults to HTTP and HTTPS to block ``file://`` and SSRF + vectors like the cloud metadata endpoint. Example: >>> from nitro_img import config @@ -40,6 +49,7 @@ class Config: """ max_input_size: int = 50_000_000 # 50 MB + max_pixels: int = 178_956_970 # ~179 Mpx (Pillow's default cap) max_output_dimensions: int = 10_000 # 10k px default_jpeg_quality: int = 85 default_webp_quality: int = 80 @@ -49,6 +59,7 @@ class Config: strip_metadata: bool = False url_timeout: float = 30.0 url_max_size: int = 50_000_000 # 50 MB + url_allowed_schemes: tuple[str, ...] = ("http", "https") def update(self, **kwargs: object) -> None: """Update one or more config fields in place. diff --git a/src/nitro_img/image.py b/src/nitro_img/image.py index 5de6f63..1651aa5 100644 --- a/src/nitro_img/image.py +++ b/src/nitro_img/image.py @@ -2,6 +2,7 @@ from __future__ import annotations +import base64 as _base64 from pathlib import Path from typing import IO @@ -11,7 +12,7 @@ from .errors import ImageFormatError from .pipeline import Pipeline from .types import Anchor, Format -from .utils import format_from_extension +from .utils import format_from_extension, mime_type from . import loaders from .operations import resize as resize_ops @@ -21,8 +22,26 @@ from .operations import overlay as overlay_ops from .operations import adjust as adjust_ops from .operations import effects as effects_ops -from .output import export from .output import optimize as optimize_mod +from .output.encode import encode as _encode_bytes + + +class _PresetProxy: + """Lazy accessor for the shared :class:`Presets` instance. + + Avoids importing ``presets`` at class-definition time so any error in + that module surfaces at the call site rather than as a class-loading + failure. + """ + + __slots__ = () + _instance: object | None = None + + def __getattr__(self, name: str) -> object: + if _PresetProxy._instance is None: + from .presets import presets as _singleton + _PresetProxy._instance = _singleton + return getattr(_PresetProxy._instance, name) class Image: @@ -50,9 +69,7 @@ class methods for bytes, base64 strings, URLs, or file-like objects. ... ) """ - # Class-level preset accessor - from .presets import Presets as _Presets - preset = _Presets() + preset = _PresetProxy() def __init__(self, source: str | Path) -> None: """Load an image from a filesystem path. @@ -65,6 +82,14 @@ def __init__(self, source: str | Path) -> None: ImageSizeError: If the file exceeds ``config.max_input_size``. """ img, fmt, path = loaders.load_from_path(source) + self._init_state(img, fmt, path) + + def _init_state( + self, + img: PILImage.Image, + fmt: Format | None, + path: str | Path | None, + ) -> None: self._original: PILImage.Image = img self._source_format: Format | None = fmt self._source_path: str | Path | None = path @@ -83,13 +108,7 @@ def _new_instance( path: str | Path | None = None, ) -> Image: inst = cls.__new__(cls) - inst._original = img - inst._source_format = fmt - inst._source_path = path - inst._pipeline = Pipeline() - inst._output_format = None - inst._output_quality = None - inst._auto_format = False + inst._init_state(img, fmt, path) return inst @classmethod @@ -165,7 +184,8 @@ def from_url(cls, url: str) -> Image: """Fetch and load an image from a URL. Requires the ``url`` extra: ``pip install nitro-image[url]``. - Download honours ``config.url_timeout`` and ``config.url_max_size``. + Download honours ``config.url_timeout``, ``config.url_max_size``, + and ``config.url_allowed_schemes``. Args: url: HTTP(S) URL pointing to an image. @@ -174,9 +194,10 @@ def from_url(cls, url: str) -> Image: A new ``Image`` instance. Raises: - ImageLoadError: If the download fails or the response is not - a valid image. - ImageSizeError: If the response exceeds ``config.url_max_size``. + ImageLoadError: If the URL scheme is rejected or the download + fails or the response is not a valid image. + ImageSizeError: If the response exceeds + ``config.url_max_size``. Example: >>> img = Image.from_url("https://example.com/photo.jpg") @@ -388,7 +409,7 @@ def mirror(self) -> Image: return self def grayscale(self) -> Image: - """Queue a grayscale conversion. + """Queue a grayscale conversion that preserves any alpha channel. Returns: The same ``Image`` for chaining. @@ -456,6 +477,10 @@ def watermark( Returns: The same ``Image`` for chaining. + Raises: + ValueError: If ``position`` is not a known anchor or + ``"tiled"``. + Example: >>> Image("photo.jpg").watermark("logo.png", position="bottom-right").save("out.jpg") """ @@ -491,6 +516,9 @@ def text_overlay( Returns: The same ``Image`` for chaining. + Raises: + ValueError: If ``position`` is not a known anchor. + Example: >>> Image("photo.jpg").text_overlay("Draft", font_size=48, color="red").save("out.jpg") """ @@ -602,6 +630,8 @@ def rounded_corners(self, radius: int) -> Image: """Queue rounded-corner masking, producing an alpha channel. Use a transparency-capable output format such as PNG or WebP. + The radius is clamped to ``min(width, height) // 2`` so corner + masks never overlap. Args: radius: Corner radius in pixels. @@ -627,9 +657,13 @@ def responsive( ) -> dict[int, bytes]: """Render the pipeline at multiple widths and return the encoded bytes. - Executes the pipeline immediately. Widths above the original size - are clamped when ``allow_upscale`` is False, which may collapse - duplicate entries in the returned dict. + Executes the pipeline once, then resizes the result to each + target width. ``widths`` are therefore measured against the + post-pipeline image, not the original source — chain + size-changing operations before ``responsive`` only when that's + the intent. Widths above the post-pipeline width are clamped + when ``allow_upscale`` is False, which may collapse duplicate + entries in the returned dict. Args: widths: Target widths in pixels. Defaults to @@ -637,7 +671,8 @@ def responsive( fmt: Output format. Falls back to the format set via ``.jpeg()``/``.webp()``/... or the source format. quality: Encoder quality applied to every width. - allow_upscale: Permit widths larger than the source width. + allow_upscale: Permit widths larger than the post-pipeline + width. Returns: Mapping of effective width to encoded bytes. @@ -651,9 +686,10 @@ def responsive( if widths is None: widths = [320, 640, 1024, 1920] out_fmt = fmt or self._output_format or self._source_format or Format.WEBP + eff_quality = quality if quality is not None else self._output_quality img = self._execute() return generate_responsive( - img, widths, fmt=out_fmt, quality=quality or self._output_quality, + img, widths, fmt=out_fmt, quality=eff_quality, allow_upscale=allow_upscale, ) @@ -670,7 +706,9 @@ def save_responsive( """Render the pipeline at multiple widths and save each to disk. Files are written as ``{name}_{width}.{ext}`` inside - ``output_dir``, which is created if it does not exist. + ``output_dir``, which is created if it does not exist. Widths + are measured against the post-pipeline image; see + :meth:`responsive` for the full caveat. Args: output_dir: Destination directory for the generated files. @@ -681,7 +719,8 @@ def save_responsive( fmt: Output format. Falls back to the format set via ``.jpeg()``/``.webp()``/... or the source format. quality: Encoder quality applied uniformly. - allow_upscale: Permit widths larger than the source width. + allow_upscale: Permit widths larger than the post-pipeline + width. Returns: Mapping of effective width to the saved ``Path``. @@ -693,12 +732,13 @@ def save_responsive( if widths is None: widths = [320, 640, 1024, 1920] out_fmt = fmt or self._output_format or self._source_format or Format.WEBP + eff_quality = quality if quality is not None else self._output_quality if name is None: name = Path(self._source_path).stem if self._source_path else "image" img = self._execute() return save_responsive( img, widths, output_dir, name, - fmt=out_fmt, quality=quality or self._output_quality, + fmt=out_fmt, quality=eff_quality, allow_upscale=allow_upscale, ) @@ -804,6 +844,9 @@ def optimize(self, target_kb: int, *, min_quality: int = 10, max_quality: int = Runs the pipeline, then repeatedly encodes at different quality levels until the output is at or below ``target_kb`` kilobytes. + When ``.auto_format()`` is in effect and no explicit format was + chosen, the format is decided per image (alpha → PNG, otherwise + WebP) before the binary search begins. Args: target_kb: Desired maximum output size in kilobytes. @@ -818,8 +861,8 @@ def optimize(self, target_kb: int, *, min_quality: int = 10, max_quality: int = >>> len(data) <= 200 * 1024 True """ - fmt = self._resolve_format() img = self._execute() + fmt = self._resolve_output_format(img) data, _ = optimize_mod.optimize( img, fmt, target_kb, min_quality=min_quality, max_quality=max_quality, ) @@ -838,7 +881,6 @@ def auto_format(self) -> Image: Example: >>> Image("photo.jpg").auto_format().save("out.webp") """ - # Deferred — we resolve at output time self._auto_format = True return self @@ -939,9 +981,33 @@ def _resolve_format(self, path: str | Path | None = None) -> Format: "or save to a path with a recognized extension." ) + def _resolve_output_format( + self, + img: PILImage.Image, + path: str | Path | None = None, + ) -> Format: + """Resolve the output format with ``auto_format`` honored.""" + if self._auto_format and self._output_format is None: + return optimize_mod.pick_format(img) + return self._resolve_format(path) + def _execute(self) -> PILImage.Image: return self._pipeline.execute(self._original.copy()) + def _encode(self, path: str | Path | None = None) -> tuple[bytes, Format]: + """Run the pipeline and encode, honoring ``auto_format``. + + Returns the encoded bytes and the format actually chosen so + callers can build content-type headers and data URIs without + re-deriving the format. + """ + img = self._execute() + if self._auto_format and self._output_format is None: + return optimize_mod.auto_format(img, quality=self._output_quality) + fmt = self._resolve_format(path) + data = _encode_bytes(img, fmt, quality=self._output_quality) + return data, fmt + def save(self, path: str | Path) -> Path: """Run the pipeline and write the result to disk. @@ -962,15 +1028,15 @@ def save(self, path: str | Path) -> Path: Example: >>> Image("photo.jpg").resize(800).save("out/photo.webp") """ - img = self._execute() - if self._auto_format and self._output_format is None: - data, fmt = optimize_mod.auto_format(img, quality=self._output_quality) - path = Path(path) + path = Path(path) + data, _ = self._encode(path) + try: path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) - return path - fmt = self._resolve_format(path) - return export.save(img, path, fmt, quality=self._output_quality) + except Exception as e: + from .errors import ImageOutputError + raise ImageOutputError(f"Cannot save to '{path}': {e}") from e + return path def to_bytes(self) -> bytes: """Run the pipeline and return the encoded image bytes. @@ -987,12 +1053,8 @@ def to_bytes(self) -> bytes: Example: >>> data = Image("photo.jpg").resize(400).webp().to_bytes() """ - img = self._execute() - if self._auto_format and self._output_format is None: - data, _ = optimize_mod.auto_format(img, quality=self._output_quality) - return data - fmt = self._resolve_format() - return export.to_bytes(img, fmt, quality=self._output_quality) + data, _ = self._encode() + return data def to_base64(self) -> str: """Run the pipeline and return a base64-encoded string. @@ -1008,13 +1070,8 @@ def to_base64(self) -> str: >>> Image("photo.jpg").resize(100).webp().to_base64() # doctest: +ELLIPSIS 'UklGR...' """ - img = self._execute() - if self._auto_format and self._output_format is None: - import base64 - data, _ = optimize_mod.auto_format(img, quality=self._output_quality) - return base64.b64encode(data).decode("ascii") - fmt = self._resolve_format() - return export.to_base64(img, fmt, quality=self._output_quality) + data, _ = self._encode() + return _base64.b64encode(data).decode("ascii") def to_data_uri(self) -> str: """Run the pipeline and return an inline data URI. @@ -1031,15 +1088,9 @@ def to_data_uri(self) -> str: >>> uri.startswith("data:image/webp;base64,") True """ - img = self._execute() - if self._auto_format and self._output_format is None: - import base64 as b64mod - from .utils import mime_type - data, fmt = optimize_mod.auto_format(img, quality=self._output_quality) - b64 = b64mod.b64encode(data).decode("ascii") - return f"data:{mime_type(fmt)};base64,{b64}" - fmt = self._resolve_format() - return export.to_data_uri(img, fmt, quality=self._output_quality) + data, fmt = self._encode() + b64 = _base64.b64encode(data).decode("ascii") + return f"data:{mime_type(fmt)};base64,{b64}" def to_response(self) -> dict: """Run the pipeline and return a framework-agnostic response dict. @@ -1056,27 +1107,23 @@ def to_response(self) -> dict: >>> resp["content_type"] 'image/webp' """ - img = self._execute() - if self._auto_format and self._output_format is None: - from .utils import mime_type - data, fmt = optimize_mod.auto_format(img, quality=self._output_quality) - return { - "body": data, - "content_type": mime_type(fmt), - "content_length": len(data), - } - fmt = self._resolve_format() - return export.to_response(img, fmt, quality=self._output_quality) + data, fmt = self._encode() + return { + "body": data, + "content_type": mime_type(fmt), + "content_length": len(data), + } # -- Framework integration responses -- def to_django_response(self, *, filename: str | None = None) -> object: """Run the pipeline and return a Django ``HttpResponse``. - Requires Django to be installed. + Requires Django to be installed. Honours ``.auto_format()`` so + the response Content-Type matches the chosen format. Args: - filename: Optional download filename; sets a + filename: Optional download filename; sets a sanitised ``Content-Disposition: inline`` header when provided. Returns: @@ -1090,18 +1137,18 @@ def to_django_response(self, *, filename: str | None = None) -> object: >>> def view(request): # doctest: +SKIP ... return Image("photo.jpg").resize(400).webp().to_django_response() """ - from .integrations import to_django_response - fmt = self._resolve_format() - img = self._execute() - return to_django_response(img, fmt, quality=self._output_quality, filename=filename) + from .integrations import _build_response_for_django + data, fmt = self._encode() + return _build_response_for_django(data, fmt, filename=filename) def to_flask_response(self, *, filename: str | None = None) -> object: """Run the pipeline and return a Flask ``Response``. - Requires Flask to be installed. + Requires Flask to be installed. Honours ``.auto_format()`` so + the response Content-Type matches the chosen format. Args: - filename: Optional download filename; sets a + filename: Optional download filename; sets a sanitised ``Content-Disposition: inline`` header when provided. Returns: @@ -1116,18 +1163,19 @@ def to_flask_response(self, *, filename: str | None = None) -> object: ... def thumb(): ... return Image("photo.jpg").resize(400).webp().to_flask_response() """ - from .integrations import to_flask_response - fmt = self._resolve_format() - img = self._execute() - return to_flask_response(img, fmt, quality=self._output_quality, filename=filename) + from .integrations import _build_response_for_flask + data, fmt = self._encode() + return _build_response_for_flask(data, fmt, filename=filename) def to_fastapi_response(self, *, filename: str | None = None) -> object: """Run the pipeline and return a FastAPI/Starlette ``Response``. Requires Starlette (installed transitively with FastAPI). + Honours ``.auto_format()`` so the response Content-Type matches + the chosen format. Args: - filename: Optional download filename; sets a + filename: Optional download filename; sets a sanitised ``Content-Disposition: inline`` header when provided. Returns: @@ -1142,10 +1190,9 @@ def to_fastapi_response(self, *, filename: str | None = None) -> object: ... async def thumb(): ... return Image("photo.jpg").resize(400).webp().to_fastapi_response() """ - from .integrations import to_fastapi_response - fmt = self._resolve_format() - img = self._execute() - return to_fastapi_response(img, fmt, quality=self._output_quality, filename=filename) + from .integrations import _build_response_for_fastapi + data, fmt = self._encode() + return _build_response_for_fastapi(data, fmt, filename=filename) # -- Info -- diff --git a/src/nitro_img/integrations.py b/src/nitro_img/integrations.py index 02587cc..b31e06a 100644 --- a/src/nitro_img/integrations.py +++ b/src/nitro_img/integrations.py @@ -1,25 +1,42 @@ -"""Framework integration helpers for Django, Flask, and FastAPI.""" +"""Framework integration helpers for Django, Flask, and FastAPI. + +The public entry points are :meth:`Image.to_django_response`, +:meth:`Image.to_flask_response`, and :meth:`Image.to_fastapi_response`. +Helpers in this module accept already-encoded bytes plus the chosen +:class:`Format` so the ``auto_format`` decision can flow through to the +response Content-Type without re-encoding the image. +""" from __future__ import annotations +from urllib.parse import quote + from .types import Format from .utils import mime_type -from .output.encode import encode -from PIL import Image as PILImage +def _content_disposition(filename: str) -> str: + """Build a safe ``Content-Disposition`` header value (RFC 6266). + + Strips control characters and quote/backslash from the ASCII + fallback, and pairs it with an RFC 5987 ``filename*`` so non-ASCII + names round-trip without breaking the header. + """ + cleaned = "".join( + ch for ch in filename if 32 <= ord(ch) < 127 and ch not in '"\\' + ) + if not cleaned: + cleaned = "image" + encoded = quote(filename, safe="") + return f'inline; filename="{cleaned}"; filename*=UTF-8\'\'{encoded}' -def to_django_response( - img: PILImage.Image, + +def _build_response_for_django( + data: bytes, fmt: Format, *, - quality: int | None = None, filename: str | None = None, ) -> object: - """Create a Django HttpResponse with the image. - - Requires Django to be installed. - """ try: from django.http import HttpResponse except ImportError: @@ -27,25 +44,18 @@ def to_django_response( "Django is required for to_django_response(). " "Install with: pip install django" ) - - data = encode(img, fmt, quality=quality) response = HttpResponse(data, content_type=mime_type(fmt)) if filename: - response["Content-Disposition"] = f'inline; filename="{filename}"' + response["Content-Disposition"] = _content_disposition(filename) return response -def to_flask_response( - img: PILImage.Image, +def _build_response_for_flask( + data: bytes, fmt: Format, *, - quality: int | None = None, filename: str | None = None, ) -> object: - """Create a Flask Response with the image. - - Requires Flask to be installed. - """ try: from flask import Response except ImportError: @@ -53,25 +63,18 @@ def to_flask_response( "Flask is required for to_flask_response(). " "Install with: pip install flask" ) - - data = encode(img, fmt, quality=quality) headers = {} if filename: - headers["Content-Disposition"] = f'inline; filename="{filename}"' + headers["Content-Disposition"] = _content_disposition(filename) return Response(data, mimetype=mime_type(fmt), headers=headers) -def to_fastapi_response( - img: PILImage.Image, +def _build_response_for_fastapi( + data: bytes, fmt: Format, *, - quality: int | None = None, filename: str | None = None, ) -> object: - """Create a FastAPI/Starlette Response with the image. - - Requires starlette to be installed (included with FastAPI). - """ try: from starlette.responses import Response except ImportError: @@ -79,9 +82,7 @@ def to_fastapi_response( "Starlette/FastAPI is required for to_fastapi_response(). " "Install with: pip install fastapi" ) - - data = encode(img, fmt, quality=quality) headers = {} if filename: - headers["Content-Disposition"] = f'inline; filename="{filename}"' + headers["Content-Disposition"] = _content_disposition(filename) return Response(content=data, media_type=mime_type(fmt), headers=headers) diff --git a/src/nitro_img/loaders.py b/src/nitro_img/loaders.py index 320d48f..91e14f6 100644 --- a/src/nitro_img/loaders.py +++ b/src/nitro_img/loaders.py @@ -6,6 +6,7 @@ import io from pathlib import Path from typing import IO, TYPE_CHECKING +from urllib.parse import urlparse from PIL import Image as PILImage @@ -24,10 +25,23 @@ def _validate_size(data: bytes) -> None: ) +def _check_pixel_count(img: PILImage.Image) -> None: + width, height = img.size + pixels = width * height + if pixels > config.max_pixels: + raise ImageSizeError( + f"Image has {pixels} pixels ({width}x{height}), exceeds limit of " + f"{config.max_pixels}. Adjust config.max_pixels to allow larger inputs." + ) + + def _post_load(img: PILImage.Image) -> PILImage.Image: if config.auto_orient: from .operations.metadata import auto_orient img = auto_orient(img) + if config.strip_metadata: + from .operations.metadata import strip_metadata + img = strip_metadata()(img) return img @@ -35,6 +49,13 @@ def _detect_format(img: PILImage.Image) -> Format | None: return format_from_pillow(img.format) +def _open_and_validate(source: object) -> PILImage.Image: + img = PILImage.open(source) + _check_pixel_count(img) + img.load() + return img + + def load_from_path(path: str | Path) -> tuple[PILImage.Image, Format | None, str | Path]: path = Path(path) if not path.exists(): @@ -45,12 +66,11 @@ def load_from_path(path: str | Path) -> tuple[PILImage.Image, Format | None, str raise ImageSizeError( f"File size {size} bytes exceeds limit of {config.max_input_size} bytes" ) - img = PILImage.open(path) - img.load() + img = _open_and_validate(path) fmt = _detect_format(img) img = _post_load(img) return img, fmt, path - except ImageSizeError: + except (ImageSizeError, ImageLoadError): raise except Exception as e: raise ImageLoadError(f"Cannot load '{path}': {e}") from e @@ -59,11 +79,12 @@ def load_from_path(path: str | Path) -> tuple[PILImage.Image, Format | None, str def load_from_bytes(data: bytes) -> tuple[PILImage.Image, Format | None, None]: _validate_size(data) try: - img = PILImage.open(io.BytesIO(data)) - img.load() + img = _open_and_validate(io.BytesIO(data)) fmt = _detect_format(img) img = _post_load(img) return img, fmt, None + except (ImageSizeError, ImageLoadError): + raise except Exception as e: raise ImageLoadError(f"Cannot load image from bytes: {e}") from e @@ -72,14 +93,13 @@ def load_from_file(file_obj: IO[bytes]) -> tuple[PILImage.Image, Format | None, try: data = file_obj.read() return load_from_bytes(data) - except ImageLoadError: + except (ImageLoadError, ImageSizeError): raise except Exception as e: raise ImageLoadError(f"Cannot load image from file object: {e}") from e def load_from_base64(b64_string: str) -> tuple[PILImage.Image, Format | None, None]: - # Strip data URI prefix if present if "," in b64_string and b64_string.startswith("data:"): b64_string = b64_string.split(",", 1)[1] try: @@ -89,16 +109,58 @@ def load_from_base64(b64_string: str) -> tuple[PILImage.Image, Format | None, No return load_from_bytes(data) +def _validate_url_scheme(url: str) -> None: + scheme = urlparse(url).scheme.lower() + allowed = tuple(s.lower() for s in config.url_allowed_schemes) + if scheme not in allowed: + raise ImageLoadError( + f"URL scheme '{scheme}' is not allowed. " + f"Permitted schemes: {', '.join(allowed)}." + ) + + +def _stream_url_body(url: str) -> bytes: + import httpx + + limit = config.url_max_size + chunks: list[bytes] = [] + total = 0 + with httpx.stream( + "GET", url, timeout=config.url_timeout, follow_redirects=True, + ) as response: + response.raise_for_status() + # Honour Content-Length when present so oversized bodies fail fast. + declared = response.headers.get("Content-Length") + if declared is not None: + try: + if int(declared) > limit: + raise ImageSizeError( + f"URL response declares {declared} bytes, exceeds limit of {limit}" + ) + except ValueError: + pass + for chunk in response.iter_bytes(): + total += len(chunk) + if total > limit: + raise ImageSizeError( + f"URL response exceeded limit of {limit} bytes; download aborted" + ) + chunks.append(chunk) + return b"".join(chunks) + + def load_from_url(url: str) -> tuple[PILImage.Image, Format | None, None]: try: - import httpx + import httpx # noqa: F401 except ImportError: raise ImageLoadError( "httpx is required for URL loading. Install with: pip install nitro-img[url]" ) + _validate_url_scheme(url) try: - response = httpx.get(url, timeout=config.url_timeout, follow_redirects=True) - response.raise_for_status() + data = _stream_url_body(url) + except (ImageSizeError, ImageLoadError): + raise except Exception as e: raise ImageLoadError(f"Cannot fetch image from '{url}': {e}") from e - return load_from_bytes(response.content) + return load_from_bytes(data) diff --git a/src/nitro_img/operations/effects.py b/src/nitro_img/operations/effects.py index c240bea..168ef00 100644 --- a/src/nitro_img/operations/effects.py +++ b/src/nitro_img/operations/effects.py @@ -43,23 +43,23 @@ def rounded_corners(radius: int) -> Callable[[PILImage.Image], PILImage.Image]: def _rounded(img: PILImage.Image) -> PILImage.Image: img = img.convert("RGBA") w, h = img.size + # Clamp the radius so corner masks never overlap or escape the canvas. + r = max(0, min(radius, w // 2, h // 2)) mask = PILImage.new("L", (w, h), 255) - draw = ImageDraw.Draw(mask) + if r == 0: + img.putalpha(mask) + return img - # Draw black corners (will become transparent) - # Top-left - draw.rectangle([0, 0, radius, radius], fill=0) - draw.pieslice([0, 0, radius * 2, radius * 2], 180, 270, fill=255) - # Top-right - draw.rectangle([w - radius, 0, w, radius], fill=0) - draw.pieslice([w - radius * 2, 0, w, radius * 2], 270, 360, fill=255) - # Bottom-left - draw.rectangle([0, h - radius, radius, h], fill=0) - draw.pieslice([0, h - radius * 2, radius * 2, h], 90, 180, fill=255) - # Bottom-right - draw.rectangle([w - radius, h - radius, w, h], fill=0) - draw.pieslice([w - radius * 2, h - radius * 2, w, h], 0, 90, fill=255) + draw = ImageDraw.Draw(mask) + draw.rectangle([0, 0, r, r], fill=0) + draw.pieslice([0, 0, r * 2, r * 2], 180, 270, fill=255) + draw.rectangle([w - r, 0, w, r], fill=0) + draw.pieslice([w - r * 2, 0, w, r * 2], 270, 360, fill=255) + draw.rectangle([0, h - r, r, h], fill=0) + draw.pieslice([0, h - r * 2, r * 2, h], 90, 180, fill=255) + draw.rectangle([w - r, h - r, w, h], fill=0) + draw.pieslice([w - r * 2, h - r * 2, w, h], 0, 90, fill=255) img.putalpha(mask) return img diff --git a/src/nitro_img/operations/metadata.py b/src/nitro_img/operations/metadata.py index 8da78fa..c61cb11 100644 --- a/src/nitro_img/operations/metadata.py +++ b/src/nitro_img/operations/metadata.py @@ -36,9 +36,20 @@ def auto_orient(img: PILImage.Image) -> PILImage.Image: def strip_metadata() -> Op: - """Remove all EXIF, IPTC, and XMP metadata.""" + """Remove all EXIF, IPTC, and XMP metadata. + + Produces a fresh image carrying only pixel data and palette (when + applicable). Does not attempt to preserve animation frames - the + output is always a single-frame image. + """ def _strip(img: PILImage.Image) -> PILImage.Image: - cleaned = PILImage.frombytes(img.mode, img.size, img.tobytes()) + if img.mode == "P": + cleaned = PILImage.new("P", img.size) + cleaned.putpalette(img.getpalette()) + cleaned.paste(img) + else: + cleaned = PILImage.new(img.mode, img.size) + cleaned.paste(img) return cleaned return _strip diff --git a/src/nitro_img/operations/overlay.py b/src/nitro_img/operations/overlay.py index edd9a06..1676652 100644 --- a/src/nitro_img/operations/overlay.py +++ b/src/nitro_img/operations/overlay.py @@ -8,18 +8,20 @@ from PIL import Image as PILImage, ImageDraw, ImageFont -_POSITION_MAP: dict[str, str] = { - "center": "center", - "top-left": "top-left", - "top-right": "top-right", - "bottom-left": "bottom-left", - "bottom-right": "bottom-right", - "top": "top", - "bottom": "bottom", - "left": "left", - "right": "right", - "tiled": "tiled", -} +_VALID_POSITIONS: frozenset[str] = frozenset({ + "center", + "top-left", "top-right", "bottom-left", "bottom-right", + "top", "bottom", "left", "right", +}) + +_VALID_WATERMARK_POSITIONS: frozenset[str] = _VALID_POSITIONS | {"tiled"} + + +def _validate_position(position: str, valid: frozenset[str]) -> None: + if position not in valid: + raise ValueError( + f"Invalid position '{position}'. Expected one of: {sorted(valid)}." + ) def _calculate_position( @@ -42,7 +44,20 @@ def _calculate_position( "left": (margin, (base_h - over_h) // 2), "right": (base_w - over_w - margin, (base_h - over_h) // 2), } - return positions.get(position, positions["bottom-right"]) + return positions[position] + + +def _load_watermark( + source: str | Path | PILImage.Image, +) -> PILImage.Image: + if isinstance(source, PILImage.Image): + wm = source.copy() + else: + wm = PILImage.open(source) + wm.load() + if wm.mode != "RGBA": + wm = wm.convert("RGBA") + return wm def watermark( @@ -53,29 +68,24 @@ def watermark( margin: int = 10, ) -> Callable[[PILImage.Image], PILImage.Image]: """Apply an image watermark at the given position with opacity.""" - def _watermark(img: PILImage.Image) -> PILImage.Image: - if isinstance(watermark_source, PILImage.Image): - wm = watermark_source.copy() - else: - wm = PILImage.open(watermark_source) + _validate_position(position, _VALID_WATERMARK_POSITIONS) - if wm.mode != "RGBA": - wm = wm.convert("RGBA") + # Load and pre-process the watermark once. Scaling that depends on the base + # image still happens inside the closure, but the disk read does not. + cached_wm = _load_watermark(watermark_source) + if opacity < 1.0: + r, g, b, a = cached_wm.split() + a = a.point(lambda x: round(x * opacity)) + cached_wm = PILImage.merge("RGBA", (r, g, b, a)) - # Scale watermark relative to base image + def _watermark(img: PILImage.Image) -> PILImage.Image: + wm = cached_wm if scale is not None: new_w = max(1, round(img.size[0] * scale)) ratio = new_w / wm.size[0] new_h = max(1, round(wm.size[1] * ratio)) wm = wm.resize((new_w, new_h), PILImage.LANCZOS) - # Apply opacity - if opacity < 1.0: - r, g, b, a = wm.split() - a = a.point(lambda x: round(x * opacity)) - wm = PILImage.merge("RGBA", (r, g, b, a)) - - # Ensure base is RGBA for compositing base = img.convert("RGBA") if img.mode != "RGBA" else img.copy() if position == "tiled": @@ -86,7 +96,6 @@ def _watermark(img: PILImage.Image) -> PILImage.Image: pos = _calculate_position(base.size, wm.size, position, margin) base.paste(wm, pos, wm) - # Convert back to original mode if needed if img.mode == "RGB": return base.convert("RGB") return base @@ -104,10 +113,11 @@ def text_overlay( margin: int = 10, ) -> Callable[[PILImage.Image], PILImage.Image]: """Add a text overlay at the given position.""" + _validate_position(position, _VALID_POSITIONS) + def _text_overlay(img: PILImage.Image) -> PILImage.Image: base = img.convert("RGBA") if img.mode != "RGBA" else img.copy() - # Create text layer txt_layer = PILImage.new("RGBA", base.size, (0, 0, 0, 0)) draw = ImageDraw.Draw(txt_layer) @@ -119,14 +129,12 @@ def _text_overlay(img: PILImage.Image) -> PILImage.Image: except OSError: font = ImageFont.load_default() - # Get text bounding box bbox = draw.textbbox((0, 0), text, font=font) text_w = bbox[2] - bbox[0] text_h = bbox[3] - bbox[1] pos = _calculate_position(base.size, (text_w, text_h), position, margin) - # Parse color and apply opacity if isinstance(color, str): from PIL import ImageColor rgba = ImageColor.getrgb(color) diff --git a/src/nitro_img/operations/resize.py b/src/nitro_img/operations/resize.py index 1081207..9bea7f2 100644 --- a/src/nitro_img/operations/resize.py +++ b/src/nitro_img/operations/resize.py @@ -6,9 +6,20 @@ from PIL import Image as PILImage +from ..config import config +from ..errors import ImageSizeError + Op = Callable[[PILImage.Image], PILImage.Image] +def _check_output_dims(width: int, height: int) -> None: + cap = config.max_output_dimensions + if width > cap or height > cap: + raise ImageSizeError( + f"Output {width}x{height} exceeds max_output_dimensions={cap}" + ) + + def resize_fit( width: int | None = None, height: int | None = None, @@ -33,7 +44,6 @@ def _resize(img: PILImage.Image) -> PILImage.Image: target_w = round(orig_w * ratio) target_h = height - # Maintain aspect ratio — fit within the box ratio_w = target_w / orig_w ratio_h = target_h / orig_h ratio = min(ratio_w, ratio_h) @@ -43,6 +53,7 @@ def _resize(img: PILImage.Image) -> PILImage.Image: new_w = max(1, round(orig_w * ratio)) new_h = max(1, round(orig_h * ratio)) + _check_output_dims(new_w, new_h) return img.resize((new_w, new_h), PILImage.LANCZOS) return _resize @@ -58,6 +69,7 @@ def thumbnail( def _thumbnail(img: PILImage.Image) -> PILImage.Image: if not allow_upscale and img.size[0] <= width and img.size[1] <= height: return img + _check_output_dims(width, height) img = img.copy() img.thumbnail((width, height), PILImage.LANCZOS) return img @@ -79,13 +91,16 @@ def _cover(img: PILImage.Image) -> PILImage.Image: ratio = max(ratio_w, ratio_h) if not allow_upscale and ratio > 1.0: - ratio = max(ratio_w, ratio_h) - if ratio > 1.0: - # Can't cover without upscaling — crop from center at original size - return _center_crop(img, min(width, orig_w), min(height, orig_h)) + # Source is smaller than target on at least one axis. Crop from the + # original at whatever size we can produce without enlarging. + crop_w = min(width, orig_w) + crop_h = min(height, orig_h) + _check_output_dims(crop_w, crop_h) + return _center_crop(img, crop_w, crop_h) new_w = max(1, round(orig_w * ratio)) new_h = max(1, round(orig_h * ratio)) + _check_output_dims(width, height) img = img.resize((new_w, new_h), PILImage.LANCZOS) return _center_crop(img, width, height) @@ -111,6 +126,7 @@ def _contain(img: PILImage.Image) -> PILImage.Image: new_w = max(1, round(orig_w * ratio)) new_h = max(1, round(orig_h * ratio)) + _check_output_dims(width, height) resized = img.resize((new_w, new_h), PILImage.LANCZOS) mode = "RGBA" if img.mode == "RGBA" else "RGB" diff --git a/src/nitro_img/operations/transform.py b/src/nitro_img/operations/transform.py index f89f15c..93a15cf 100644 --- a/src/nitro_img/operations/transform.py +++ b/src/nitro_img/operations/transform.py @@ -31,7 +31,12 @@ def _mirror(img: PILImage.Image) -> PILImage.Image: def grayscale() -> Op: - """Convert the image to grayscale.""" + """Convert the image to grayscale, preserving any alpha channel.""" def _grayscale(img: PILImage.Image) -> PILImage.Image: + if img.mode == "RGBA": + alpha = img.getchannel("A") + gray = img.convert("L").convert("RGB").convert("RGBA") + gray.putalpha(alpha) + return gray return img.convert("L").convert("RGB") return _grayscale diff --git a/src/nitro_img/output/optimize.py b/src/nitro_img/output/optimize.py index 83affd4..c71915e 100644 --- a/src/nitro_img/output/optimize.py +++ b/src/nitro_img/output/optimize.py @@ -54,6 +54,27 @@ def optimize( return data, min_quality +def _has_meaningful_alpha(img: PILImage.Image) -> bool: + if img.mode not in ("RGBA", "LA", "PA"): + return False + if img.mode == "RGBA": + extrema = img.getchannel("A").getextrema() + return extrema != (255, 255) + return True + + +def pick_format(img: PILImage.Image) -> Format: + """Pick a sensible output format for ``img`` without encoding twice. + + Mirrors :func:`auto_format`'s decision but skips the WebP/JPEG size + comparison so callers like :func:`optimize` can run their own quality + search against a single format. + """ + if _has_meaningful_alpha(img): + return Format.PNG + return Format.WEBP + + def auto_format(img: PILImage.Image, quality: int | None = None) -> tuple[bytes, Format]: """Pick the best format for the image content. @@ -61,21 +82,10 @@ def auto_format(img: PILImage.Image, quality: int | None = None) -> tuple[bytes, - Photographic content -> WebP (smaller than JPEG) - Fallback -> WebP """ - has_alpha = img.mode in ("RGBA", "LA", "PA") - - if has_alpha: - # Check if alpha channel is actually used - if img.mode == "RGBA": - alpha = img.getchannel("A") - extrema = alpha.getextrema() - if extrema == (255, 255): - has_alpha = False # Fully opaque, alpha not needed - - if has_alpha: + if _has_meaningful_alpha(img): data = encode(img, Format.PNG) return data, Format.PNG - # Compare WebP vs JPEG size at equivalent quality webp_data = encode(img, Format.WEBP, quality=quality) jpeg_data = encode(img, Format.JPEG, quality=quality) diff --git a/src/nitro_img/pipeline.py b/src/nitro_img/pipeline.py index 03b8c2e..49a9c67 100644 --- a/src/nitro_img/pipeline.py +++ b/src/nitro_img/pipeline.py @@ -2,32 +2,35 @@ from __future__ import annotations -from dataclasses import dataclass, field -from typing import Any, Callable +from dataclasses import dataclass +from typing import Callable from PIL import Image as PILImage -from .errors import ImageProcessingError +from .errors import ImageProcessingError, NitroImgError @dataclass class Operation: name: str fn: Callable[[PILImage.Image], PILImage.Image] - kwargs: dict[str, Any] = field(default_factory=dict) class Pipeline: def __init__(self) -> None: self._operations: list[Operation] = [] - def add(self, name: str, fn: Callable[[PILImage.Image], PILImage.Image], **kwargs: Any) -> None: - self._operations.append(Operation(name=name, fn=fn, kwargs=kwargs)) + def add(self, name: str, fn: Callable[[PILImage.Image], PILImage.Image]) -> None: + self._operations.append(Operation(name=name, fn=fn)) def execute(self, img: PILImage.Image) -> PILImage.Image: for op in self._operations: try: img = op.fn(img) + except NitroImgError: + # Library-level errors (size caps, validation, etc.) carry + # their own meaning - don't disguise them as pipeline failures. + raise except Exception as e: raise ImageProcessingError( f"Operation '{op.name}' failed: {e}" diff --git a/src/nitro_img/placeholder.py b/src/nitro_img/placeholder.py index 3edb981..82ebf3f 100644 --- a/src/nitro_img/placeholder.py +++ b/src/nitro_img/placeholder.py @@ -26,38 +26,33 @@ def lqip(img: PILImage.Image, width: int = 20) -> str: return f"data:image/webp;base64,{b64}" -def dominant_color(img: PILImage.Image, sample_size: int = 100) -> str: - """Extract the dominant color as a hex string (#RRGGBB). - - Downscales the image then finds the most common color bucket. - """ +def _quantized_color_counts(img: PILImage.Image, sample_size: int) -> Counter: small = img.copy() small.thumbnail((sample_size, sample_size), PILImage.LANCZOS) small = small.convert("RGB") - pixels = list(zip(small.tobytes()[0::3], small.tobytes()[1::3], small.tobytes()[2::3])) - # Quantize to reduce color space (bucket to nearest 16) - quantized = [ + raw = small.tobytes() + pixels = zip(raw[0::3], raw[1::3], raw[2::3]) + quantized = ( ((r >> 4) << 4, (g >> 4) << 4, (b >> 4) << 4) for r, g, b in pixels - ] - counter = Counter(quantized) + ) + return Counter(quantized) + + +def dominant_color(img: PILImage.Image, sample_size: int = 100) -> str: + """Extract the dominant color as a hex string (#RRGGBB). + + Downscales the image then finds the most common color bucket. + """ + counter = _quantized_color_counts(img, sample_size) most_common = counter.most_common(1)[0][0] return f"#{most_common[0]:02x}{most_common[1]:02x}{most_common[2]:02x}" def color_palette(img: PILImage.Image, count: int = 5, sample_size: int = 100) -> list[str]: """Extract a color palette as a list of hex strings.""" - small = img.copy() - small.thumbnail((sample_size, sample_size), PILImage.LANCZOS) - small = small.convert("RGB") - - pixels = list(zip(small.tobytes()[0::3], small.tobytes()[1::3], small.tobytes()[2::3])) - quantized = [ - ((r >> 4) << 4, (g >> 4) << 4, (b >> 4) << 4) - for r, g, b in pixels - ] - counter = Counter(quantized) + counter = _quantized_color_counts(img, sample_size) return [ f"#{r:02x}{g:02x}{b:02x}" for (r, g, b), _ in counter.most_common(count) diff --git a/src/nitro_img/utils.py b/src/nitro_img/utils.py index d0f63e3..5e0a79b 100644 --- a/src/nitro_img/utils.py +++ b/src/nitro_img/utils.py @@ -37,6 +37,15 @@ Format.TIFF: "image/tiff", } +_FORMAT_TO_EXT: dict[Format, str] = { + Format.JPEG: ".jpg", + Format.PNG: ".png", + Format.WEBP: ".webp", + Format.GIF: ".gif", + Format.BMP: ".bmp", + Format.TIFF: ".tiff", +} + def format_from_extension(path: str | Path) -> Format | None: ext = os.path.splitext(str(path))[1].lower() @@ -54,7 +63,4 @@ def mime_type(fmt: Format) -> str: def extension_for_format(fmt: Format) -> str: - for ext, f in _EXT_TO_FORMAT.items(): - if f == fmt: - return ext - return ".bin" + return _FORMAT_TO_EXT.get(fmt, ".bin") diff --git a/tests/test_config.py b/tests/test_config.py new file mode 100644 index 0000000..e81cb60 --- /dev/null +++ b/tests/test_config.py @@ -0,0 +1,167 @@ +"""Tests for the global config singleton and its enforcement.""" + +from __future__ import annotations + +import pytest +from PIL import Image as PILImage + +from nitro_img import Image, ImageSizeError, config + + +@pytest.fixture(autouse=True) +def restore_config(): + snapshot = { + "max_input_size": config.max_input_size, + "max_pixels": config.max_pixels, + "max_output_dimensions": config.max_output_dimensions, + "strip_metadata": config.strip_metadata, + "auto_orient": config.auto_orient, + "url_allowed_schemes": config.url_allowed_schemes, + "url_max_size": config.url_max_size, + } + yield + for key, value in snapshot.items(): + setattr(config, key, value) + + +class TestConfigUpdate: + def test_update_known_field(self): + config.update(default_jpeg_quality=99) + assert config.default_jpeg_quality == 99 + config.update(default_jpeg_quality=85) + + def test_update_unknown_field_raises(self): + with pytest.raises(ValueError, match="Unknown config option"): + config.update(does_not_exist=True) + + def test_update_multiple_fields(self): + config.update(default_jpeg_quality=70, default_webp_quality=70) + assert config.default_jpeg_quality == 70 + assert config.default_webp_quality == 70 + + +class TestMaxPixels: + def test_load_above_max_pixels_raises(self, sample_jpg): + config.update(max_pixels=100) # 800x600 = 480_000 — well over 100 + with pytest.raises(ImageSizeError, match="pixels"): + Image(sample_jpg) + + def test_load_below_max_pixels_succeeds(self, sample_jpg): + config.update(max_pixels=10_000_000) + img = Image(sample_jpg) + assert img.width == 800 + + +class TestMaxOutputDimensions: + def test_resize_above_cap_raises(self, sample_jpg): + config.update(max_output_dimensions=400) + with pytest.raises(ImageSizeError, match="max_output_dimensions"): + Image(sample_jpg).resize(800, allow_upscale=True).to_bytes() + + def test_cover_above_cap_raises(self, sample_jpg): + config.update(max_output_dimensions=200) + with pytest.raises(ImageSizeError, match="max_output_dimensions"): + Image(sample_jpg).cover(400, 400).to_bytes() + + def test_contain_above_cap_raises(self, sample_jpg): + config.update(max_output_dimensions=200) + with pytest.raises(ImageSizeError, match="max_output_dimensions"): + Image(sample_jpg).contain(400, 400).to_bytes() + + def test_thumbnail_above_cap_raises(self, sample_jpg): + config.update(max_output_dimensions=100) + # thumbnail only runs when the image exceeds the box, so allow upscale path + with pytest.raises(ImageSizeError): + Image(sample_jpg).thumbnail(400, 400, allow_upscale=True).to_bytes() + + def test_within_cap_succeeds(self, sample_jpg): + config.update(max_output_dimensions=2000) + Image(sample_jpg).resize(400).webp().to_bytes() + + +class TestStripMetadataConfig: + def test_strip_on_load_removes_exif(self, tmp_path): + path = tmp_path / "withexif.jpg" + pil = PILImage.new("RGB", (50, 50), (10, 20, 30)) + exif = pil.getexif() + exif[0x0112] = 6 # Orientation tag with rotation + pil.save(path, "JPEG", exif=exif) + + config.update(strip_metadata=True, auto_orient=False) + img = Image(path) + meta = img.get_metadata() + # After stripping, EXIF should be empty or absent + assert not meta.get("exif") + + def test_no_strip_when_disabled(self, tmp_path): + path = tmp_path / "withexif.jpg" + pil = PILImage.new("RGB", (50, 50), (10, 20, 30)) + exif = pil.getexif() + exif[0x010E] = "test caption" # ImageDescription tag + pil.save(path, "JPEG", exif=exif) + + config.update(strip_metadata=False, auto_orient=False) + img = Image(path) + meta = img.get_metadata() + assert meta.get("exif") + + +class TestAutoOrient: + def _save_oriented(self, path, orientation: int) -> None: + # Build a tall image with distinct top/bottom colors so orientation + # transforms are detectable. + pil = PILImage.new("RGB", (40, 60), (255, 0, 0)) + # Draw a blue strip at the bottom 1/3 so transforms move it predictably. + bottom = PILImage.new("RGB", (40, 20), (0, 0, 255)) + pil.paste(bottom, (0, 40)) + exif = pil.getexif() + exif[0x0112] = orientation + pil.save(path, "JPEG", exif=exif) + + def test_auto_orient_applies_rotation(self, tmp_path): + path = tmp_path / "rot.jpg" + self._save_oriented(path, orientation=6) # 90° CW + config.update(auto_orient=True) + img = Image(path) + # Orientation 6 swaps width/height (60x40 after rotation) + assert img.width == 60 + assert img.height == 40 + + def test_auto_orient_disabled_keeps_raw_dims(self, tmp_path): + path = tmp_path / "rot.jpg" + self._save_oriented(path, orientation=6) + config.update(auto_orient=False) + img = Image(path) + assert img.width == 40 + assert img.height == 60 + + def test_auto_orient_no_exif_is_safe(self, sample_jpg): + config.update(auto_orient=True) + img = Image(sample_jpg) + assert img.width == 800 + + +class TestMaxInputSize: + def test_oversized_path_raises(self, sample_jpg): + config.update(max_input_size=10) + with pytest.raises(ImageSizeError, match="File size"): + Image(sample_jpg) + + def test_oversized_bytes_raises(self, sample_bytes): + config.update(max_input_size=10) + with pytest.raises(ImageSizeError, match="Input size"): + Image.from_bytes(sample_bytes) + + +class TestUrlSchemeAllowlist: + def test_disallowed_scheme_raises(self): + with pytest.raises(Exception) as exc: + Image.from_url("file:///etc/passwd") + # Either an ImportError (httpx missing) or ImageLoadError (scheme rejected). + msg = str(exc.value).lower() + assert "scheme" in msg or "httpx" in msg + + def test_default_allowed_schemes(self): + assert "http" in config.url_allowed_schemes + assert "https" in config.url_allowed_schemes + assert "file" not in config.url_allowed_schemes diff --git a/tests/test_from_url.py b/tests/test_from_url.py new file mode 100644 index 0000000..0301136 --- /dev/null +++ b/tests/test_from_url.py @@ -0,0 +1,118 @@ +"""Tests for Image.from_url and the URL streaming loader. + +These tests stub httpx so no real network calls are made. +""" + +from __future__ import annotations + +import io +import sys +from types import SimpleNamespace + +import pytest +from PIL import Image as PILImage + +from nitro_img import Image, ImageLoadError, ImageSizeError, config + + +class _FakeResponse: + def __init__(self, body: bytes, status_code: int = 200, headers: dict | None = None): + self._body = body + self.status_code = status_code + self.headers = headers or {} + + def raise_for_status(self) -> None: + if self.status_code >= 400: + raise RuntimeError(f"HTTP {self.status_code}") + + def iter_bytes(self): + # Yield in two chunks so streaming behavior is exercised. + mid = len(self._body) // 2 + if mid > 0: + yield self._body[:mid] + yield self._body[mid:] + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + +def _png_bytes(w: int = 40, h: int = 30) -> bytes: + img = PILImage.new("RGB", (w, h), (10, 20, 30)) + buf = io.BytesIO() + img.save(buf, "PNG") + return buf.getvalue() + + +@pytest.fixture +def fake_httpx(monkeypatch): + body_ref = {"body": _png_bytes(), "headers": {}} + + def fake_stream(method, url, **kwargs): + return _FakeResponse(body_ref["body"], headers=body_ref["headers"]) + + fake = SimpleNamespace(stream=fake_stream) + monkeypatch.setitem(sys.modules, "httpx", fake) + return body_ref + + +class TestFromUrlHappyPath: + def test_loads_png_from_url(self, fake_httpx): + img = Image.from_url("https://example.com/photo.png") + assert img.width == 40 + assert img.height == 30 + + def test_streaming_aggregates_chunks(self, fake_httpx): + fake_httpx["body"] = _png_bytes(120, 80) + img = Image.from_url("https://example.com/big.png") + assert img.size == (120, 80) + + +class TestFromUrlSchemeAllowlist: + def test_file_scheme_rejected(self, fake_httpx): + with pytest.raises(ImageLoadError, match="scheme"): + Image.from_url("file:///etc/passwd") + + def test_ftp_scheme_rejected(self, fake_httpx): + with pytest.raises(ImageLoadError, match="scheme"): + Image.from_url("ftp://example.com/photo.jpg") + + def test_custom_allowlist(self, fake_httpx, monkeypatch): + original = config.url_allowed_schemes + config.update(url_allowed_schemes=("http",)) + try: + with pytest.raises(ImageLoadError, match="scheme"): + Image.from_url("https://example.com/photo.png") + finally: + config.update(url_allowed_schemes=original) + + +class TestFromUrlSizeCap: + def test_streaming_aborts_when_over_cap(self, fake_httpx): + original = config.url_max_size + config.update(url_max_size=10) + try: + with pytest.raises(ImageSizeError, match="aborted|exceeds"): + Image.from_url("https://example.com/photo.png") + finally: + config.update(url_max_size=original) + + def test_content_length_header_short_circuits(self, fake_httpx): + original = config.url_max_size + config.update(url_max_size=50) + try: + fake_httpx["headers"] = {"Content-Length": "10000"} + with pytest.raises(ImageSizeError, match="declares"): + Image.from_url("https://example.com/photo.png") + finally: + config.update(url_max_size=original) + + +class TestFromUrlHttpxMissing: + def test_missing_httpx_raises_clear_error(self, monkeypatch): + # Make the import fail. + monkeypatch.setitem(sys.modules, "httpx", None) + with pytest.raises(ImageLoadError, match="httpx"): + Image.from_url("https://example.com/photo.png") diff --git a/tests/test_review_fixes.py b/tests/test_review_fixes.py new file mode 100644 index 0000000..987ce78 --- /dev/null +++ b/tests/test_review_fixes.py @@ -0,0 +1,186 @@ +"""Regression tests covering issues from the code review. + +Each test class targets a specific finding so future drift surfaces in +its own failure rather than mingled with unrelated suites. +""" + +from __future__ import annotations + +import pytest +from PIL import Image as PILImage + +from nitro_img import BatchImage, Image +from nitro_img.integrations import _content_disposition + + +class TestAutoFormatHonoredByOptimize: + def test_auto_format_then_optimize_runs(self, sample_jpg): + data = Image(sample_jpg).auto_format().optimize(target_kb=200) + assert isinstance(data, bytes) + assert len(data) > 0 + + def test_auto_format_chooses_png_for_rgba(self, sample_png): + # PNG fixture is RGBA — optimize should pick PNG, which has no + # quality knob so the binary search short-circuits. + data = Image(sample_png).auto_format().optimize(target_kb=200) + assert data.startswith(b"\x89PNG\r\n") + + +class TestQualityZeroNotFalsy: + def test_quality_zero_in_responsive(self, sample_jpg): + # quality=0 used to be swallowed by `or self._output_quality`. + # Now it should pass through and produce visibly heavier compression + # than quality=90. + zero = Image(sample_jpg).resize(400).webp().responsive([200], quality=0) + ninety = Image(sample_jpg).resize(400).webp().responsive([200], quality=90) + assert sum(len(b) for b in zero.values()) <= sum(len(b) for b in ninety.values()) + + +class TestPositionValidation: + def test_invalid_watermark_position_raises(self, sample_jpg, sample_png): + with pytest.raises(ValueError, match="position"): + Image(sample_jpg).watermark(sample_png, position="nope") + + def test_invalid_text_position_raises(self, sample_jpg): + with pytest.raises(ValueError, match="position"): + Image(sample_jpg).text_overlay("hi", position="middle") + + def test_tiled_position_allowed_for_watermark(self, sample_jpg, sample_png): + # Should NOT raise — tiled is only valid for image watermarks. + Image(sample_jpg).watermark(sample_png, position="tiled").webp().to_bytes() + + def test_tiled_position_rejected_for_text(self, sample_jpg): + with pytest.raises(ValueError, match="position"): + Image(sample_jpg).text_overlay("hi", position="tiled") + + +class TestBatchImageGif: + def test_gif_recorder_present(self, tmp_path, sample_jpg, sample_png): + # Create a batch with two images + src = tmp_path / "src" + src.mkdir() + Image(sample_jpg).save(src / "a.png") + Image(sample_png).save(src / "b.png") + + out_dir = tmp_path / "out" + paths = ( + BatchImage(str(src / "*.png")) + .resize(50) + .gif() + .save(str(out_dir / "{name}.gif")) + ) + assert len(paths) == 2 + for p in paths: + assert p.suffix == ".gif" + + +class TestContentDispositionSanitisation: + def test_strips_quote_and_backslash(self): + header = _content_disposition('evil".bad\\filename.jpg') + assert '"' not in header.split('filename*=')[0].replace('filename="', '"x', 1).replace('"', '', 1) or True + # Easier: verify there's no injection of an unmatched quote + # Count quotes in the filename= section + section = header.split('filename*=')[0] + # Should be exactly 2 quotes wrapping the filename + assert section.count('"') == 2 + + def test_strips_newlines(self): + header = _content_disposition("photo.jpg\r\nX-Injected: yes") + assert "\n" not in header + assert "\r" not in header + + def test_fallback_when_all_chars_stripped(self): + header = _content_disposition("\x00\x01\x02") + assert 'filename="image"' in header + + def test_unicode_in_filename_star(self): + header = _content_disposition("café.jpg") + # RFC 5987 percent-encoded value + assert "filename*=UTF-8''" in header + assert "caf%C3%A9.jpg" in header + + +class TestRoundedCornersClamp: + def test_huge_radius_does_not_crash(self, sample_jpg): + # radius > min dim should clamp rather than producing broken output + data = Image(sample_jpg).rounded_corners(10_000).png().to_bytes() + assert data.startswith(b"\x89PNG\r\n") + + def test_zero_radius_is_noop_alpha(self, sample_jpg): + # radius 0 should still produce a valid RGBA image + data = Image(sample_jpg).rounded_corners(0).png().to_bytes() + assert data.startswith(b"\x89PNG\r\n") + + +class TestGrayscalePreservesAlpha: + def test_rgba_input_keeps_alpha(self, sample_png): + from PIL import Image as PILImage + + import io as _io + + data = Image(sample_png).grayscale().png().to_bytes() + result = PILImage.open(_io.BytesIO(data)) + assert result.mode == "RGBA" + + +class TestStripMetadataPaletteMode: + def test_palette_image_survives_strip(self, tmp_path): + # Create a P-mode image + path = tmp_path / "palette.png" + pil = PILImage.new("P", (40, 30)) + pil.putpalette([i % 256 for i in range(768)]) + pil.save(path, "PNG") + + # Stripping should not raise and should produce valid output + data = Image(path).strip_metadata().png().to_bytes() + assert data.startswith(b"\x89PNG\r\n") + + +class TestExtensionForFormat: + def test_jpeg_extension_is_jpg(self): + from nitro_img import Format + from nitro_img.utils import extension_for_format + + assert extension_for_format(Format.JPEG) == ".jpg" + assert extension_for_format(Format.TIFF) == ".tiff" + + +class TestFrameworkResponseAutoFormat: + """Ensure auto_format flows through framework response helpers. + + We can't import Django here, but we can prove the encode/format + decision happens before the framework import error by checking that + ImageFormatError isn't raised anymore on bytes-loaded images. + """ + + def test_django_response_with_auto_format_does_not_raise_format_error(self, sample_jpg): + # Django isn't installed; if auto_format works, we should hit + # the ImportError ("Django"), not ImageFormatError. + with pytest.raises(ImportError, match="Django"): + Image.from_bytes(open(sample_jpg, "rb").read()).auto_format().to_django_response() + + def test_flask_response_with_auto_format_does_not_raise_format_error(self, sample_jpg): + with pytest.raises(ImportError, match="Flask"): + Image.from_bytes(open(sample_jpg, "rb").read()).auto_format().to_flask_response() + + def test_fastapi_response_with_auto_format_does_not_raise_format_error(self, sample_jpg): + with pytest.raises(ImportError, match="Starlette"): + Image.from_bytes(open(sample_jpg, "rb").read()).auto_format().to_fastapi_response() + + +class TestWatermarkCaching: + def test_watermark_works_when_file_deleted_after_chain(self, tmp_path, sample_jpg): + # Build a watermark, kick off a pipeline, delete the source, then + # call save_responsive which triggers the pipeline multiple times. + # If the watermark were re-opened per execution this would fail. + wm_path = tmp_path / "wm.png" + wm = PILImage.new("RGBA", (30, 30), (255, 255, 255, 200)) + wm.save(wm_path, "PNG") + + img = Image(sample_jpg).watermark(str(wm_path), opacity=0.5) + # Delete the watermark before any output runs + wm_path.unlink() + # Now produce outputs — should still succeed thanks to closure caching + out_dir = tmp_path / "out" + paths = img.webp().save_responsive(out_dir, [100, 200]) + assert len(paths) >= 1 From 9600ae2ca126ee0b3763801b4825be6fa50125ec Mon Sep 17 00:00:00 2001 From: Sean N Date: Sat, 16 May 2026 09:21:14 +0200 Subject: [PATCH 2/2] Updating the project description --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index f465f13..5718df3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -5,7 +5,7 @@ build-backend = "hatchling.build" [project] name = "nitro-image" version = "1.1.1" -description = "Fast, friendly image processing for web apps and SaaS" +description = "Resize, convert, and optimize images with a chainable API. Generate responsive sizes in one call." readme = "README.md" license = "BSD-3-Clause" requires-python = ">=3.10"