From 68ec3d18301f3eb4a5f6af391175bf981204772b Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sat, 13 Jun 2026 01:00:06 -0700 Subject: [PATCH 1/2] feat(support): add Uri fluent parser with immutable Uriable builder MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements Uri.of(url) → Uriable fluent API modelled on the existing Str/Stringable pattern. Uriable wraps urllib.parse.ParseResult and exposes read accessors (scheme, host, port, path, query, fragment) alongside immutable mutators (with_scheme, with_host, with_port, with_path, append_path, with_query, add_query, remove_query, with_fragment, without_fragment, without_query). Every mutator returns a new Uriable so the original is never modified. Exports Uri and Uriable from fastapi_startkit.support. 48 new tests added in tests/utils/test_uri.py covering accessors, each mutator, fluent chaining, immutability, equality, and __repr__. Closes #178 Co-Authored-By: Claude Sonnet 4.6 --- .../src/fastapi_startkit/support/__init__.py | 3 +- .../src/fastapi_startkit/support/uri.py | 220 +++++++++++++++ fastapi_startkit/tests/utils/test_uri.py | 267 ++++++++++++++++++ 3 files changed, 489 insertions(+), 1 deletion(-) create mode 100644 fastapi_startkit/src/fastapi_startkit/support/uri.py create mode 100644 fastapi_startkit/tests/utils/test_uri.py diff --git a/fastapi_startkit/src/fastapi_startkit/support/__init__.py b/fastapi_startkit/src/fastapi_startkit/support/__init__.py index e8f3e426..683a7c08 100644 --- a/fastapi_startkit/src/fastapi_startkit/support/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/support/__init__.py @@ -1,4 +1,5 @@ from .collection import Collection, collect from .string import Str, Stringable +from .uri import Uri, Uriable -__all__ = ["Collection", "collect", "Str", "Stringable"] +__all__ = ["Collection", "collect", "Str", "Stringable", "Uri", "Uriable"] diff --git a/fastapi_startkit/src/fastapi_startkit/support/uri.py b/fastapi_startkit/src/fastapi_startkit/support/uri.py new file mode 100644 index 00000000..59e6193a --- /dev/null +++ b/fastapi_startkit/src/fastapi_startkit/support/uri.py @@ -0,0 +1,220 @@ +"""Fluent URI parser and builder — fastapi_startkit.support.uri. + +Usage:: + + from fastapi_startkit.support.uri import Uri + + url = ( + Uri.of("https://example.com/api/users?page=1&sort=asc") + .with_path("/api/posts") + .add_query("page", 2) + .remove_query("sort") + .with_fragment("results") + .get() + ) + # "https://example.com/api/posts?page=2#results" +""" + +from __future__ import annotations + +from typing import Any +from urllib.parse import ParseResult, parse_qs, urlencode, urlparse, urlunparse + + +class Uriable: + """Fluent URI builder returned by :meth:`Uri.of`. + + Every mutating method returns a *new* ``Uriable`` so the original is + not modified (value-object semantics). + """ + + def __init__( + self, + parsed: ParseResult, + query: dict[str, list[str]] | None = None, + ) -> None: + self._parsed = parsed + # Maintain query params as ``dict[str, list[str]]`` — same format as + # ``urllib.parse.parse_qs``. We keep this separately so round-tripping + # through urlencode/parse_qs stays consistent. + self._query: dict[str, list[str]] = ( + query if query is not None else parse_qs(parsed.query, keep_blank_values=True) + ) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _build_netloc(self, hostname: str | None = None, port: int | None = ...) -> str: # type: ignore[assignment] + """Reconstruct the ``netloc`` component preserving auth information.""" + hostname = hostname if hostname is not None else (self._parsed.hostname or "") + port = self._parsed.port if port is ... else port + + netloc = hostname + if port: + netloc = f"{netloc}:{port}" + + username = self._parsed.username + password = self._parsed.password + if username: + auth = f"{username}:{password}@" if password else f"{username}@" + netloc = auth + netloc + + return netloc + + def _rebuild(self, **overrides: Any) -> Uriable: + """Return a new ``Uriable`` with selective component overrides.""" + new_query: dict[str, list[str]] = overrides.pop("query", self._query) + + new_parsed = ParseResult( + scheme=overrides.get("scheme", self._parsed.scheme), + netloc=overrides.get("netloc", self._parsed.netloc), + path=overrides.get("path", self._parsed.path), + params=overrides.get("params", self._parsed.params), + query=urlencode(new_query, doseq=True), + fragment=overrides.get("fragment", self._parsed.fragment), + ) + return Uriable(new_parsed, new_query) + + # ------------------------------------------------------------------ + # Read accessors + # ------------------------------------------------------------------ + + def scheme(self) -> str: + """Return the URI scheme (e.g. ``'https'``).""" + return self._parsed.scheme + + def host(self) -> str: + """Return the hostname without port (e.g. ``'example.com'``).""" + return self._parsed.hostname or "" + + def port(self) -> int | None: + """Return the port number, or ``None`` if not explicitly set.""" + return self._parsed.port + + def path(self) -> str: + """Return the path component (e.g. ``'/api/users'``).""" + return self._parsed.path + + def query(self) -> dict[str, list[str]]: + """Return all query parameters as ``{key: [value, ...]}``.""" + return dict(self._query) + + def query_param(self, key: str, default: Any = None) -> str | None: + """Return the *first* value for ``key``, or *default* if absent.""" + values = self._query.get(key) + return values[0] if values else default + + def fragment(self) -> str: + """Return the fragment (hash) component, without the leading ``#``.""" + return self._parsed.fragment + + def get(self) -> str: + """Return the fully assembled URI string.""" + return urlunparse(self._parsed) + + # ------------------------------------------------------------------ + # Fluent mutators — each returns a new Uriable + # ------------------------------------------------------------------ + + def with_scheme(self, scheme: str) -> Uriable: + """Replace the scheme (e.g. ``'http'`` → ``'https'``).""" + return self._rebuild(scheme=scheme) + + def with_host(self, host: str) -> Uriable: + """Replace the hostname, preserving port and authentication info.""" + netloc = self._build_netloc(hostname=host) + return self._rebuild(netloc=netloc) + + def with_port(self, port: int | None) -> Uriable: + """Set (or clear) the port, preserving hostname and auth.""" + netloc = self._build_netloc(port=port) + return self._rebuild(netloc=netloc) + + def with_path(self, path: str) -> Uriable: + """Replace the entire path component.""" + return self._rebuild(path=path) + + def append_path(self, segment: str) -> Uriable: + """Append *segment* to the current path, handling trailing slashes.""" + base = self._parsed.path.rstrip("/") + segment = segment.lstrip("/") + return self._rebuild(path=f"{base}/{segment}") + + def with_query(self, params: dict[str, Any]) -> Uriable: + """Replace *all* query parameters with the supplied mapping.""" + new_query: dict[str, list[str]] = {} + for key, value in params.items(): + if isinstance(value, (list, tuple)): + new_query[key] = [str(v) for v in value] + else: + new_query[key] = [str(value)] + return self._rebuild(query=new_query) + + def add_query(self, key: str, value: Any) -> Uriable: + """Add or replace a single query parameter.""" + new_query = dict(self._query) + new_query[key] = [str(value)] + return self._rebuild(query=new_query) + + def remove_query(self, key: str) -> Uriable: + """Remove a query parameter (no-op if absent).""" + new_query = {k: v for k, v in self._query.items() if k != key} + return self._rebuild(query=new_query) + + def with_fragment(self, fragment: str) -> Uriable: + """Set the fragment (hash), without the leading ``#``.""" + return self._rebuild(fragment=fragment) + + def without_fragment(self) -> Uriable: + """Remove the fragment component.""" + return self._rebuild(fragment="") + + def without_query(self) -> Uriable: + """Remove all query parameters.""" + return self._rebuild(query={}) + + # ------------------------------------------------------------------ + # Dunder helpers + # ------------------------------------------------------------------ + + def __str__(self) -> str: + return self.get() + + def __repr__(self) -> str: + return f"Uriable({self.get()!r})" + + def __eq__(self, other: object) -> bool: + if isinstance(other, Uriable): + return self.get() == other.get() + if isinstance(other, str): + return self.get() == other + return NotImplemented + + +class Uri: + """Static factory and utility for fluent URI manipulation. + + Example:: + + from fastapi_startkit.support.uri import Uri + + uri = Uri.of("https://api.example.com/v1/users?active=true") + new_uri = ( + uri.append_path("search") + .add_query("q", "alice") + .with_scheme("http") + .get() + ) + # "http://api.example.com/v1/users/search?active=true&q=alice" + """ + + @classmethod + def of(cls, url: str) -> Uriable: + """Parse *url* and return a fluent :class:`Uriable` builder.""" + return Uriable(urlparse(url)) + + @classmethod + def parse(cls, url: str) -> Uriable: + """Alias for :meth:`of`.""" + return cls.of(url) diff --git a/fastapi_startkit/tests/utils/test_uri.py b/fastapi_startkit/tests/utils/test_uri.py new file mode 100644 index 00000000..a3925b0b --- /dev/null +++ b/fastapi_startkit/tests/utils/test_uri.py @@ -0,0 +1,267 @@ +"""Tests for the Uri fluent parser (task #178).""" + +import pytest + +from fastapi_startkit.support.uri import Uri, Uriable + + +class TestUriOf: + def test_returns_uriable(self): + result = Uri.of("https://example.com") + assert isinstance(result, Uriable) + + def test_parse_alias(self): + result = Uri.parse("https://example.com/path") + assert isinstance(result, Uriable) + assert result.path() == "/path" + + def test_get_returns_original_url(self): + url = "https://example.com/path?q=1#frag" + assert Uri.of(url).get() == url + + def test_str_returns_url(self): + url = "https://example.com/path" + assert str(Uri.of(url)) == url + + +class TestUriAccessors: + def setup_method(self): + self.uri = Uri.of("https://user:pass@example.com:8080/api/v1?page=2&sort=asc#top") + + def test_scheme(self): + assert self.uri.scheme() == "https" + + def test_host(self): + assert self.uri.host() == "example.com" + + def test_port(self): + assert self.uri.port() == 8080 + + def test_path(self): + assert self.uri.path() == "/api/v1" + + def test_fragment(self): + assert self.uri.fragment() == "top" + + def test_query_returns_dict(self): + q = self.uri.query() + assert q["page"] == ["2"] + assert q["sort"] == ["asc"] + + def test_query_param_single(self): + assert self.uri.query_param("page") == "2" + assert self.uri.query_param("sort") == "asc" + + def test_query_param_missing_returns_default(self): + assert self.uri.query_param("missing") is None + assert self.uri.query_param("missing", "fallback") == "fallback" + + def test_no_port_returns_none(self): + uri = Uri.of("https://example.com/path") + assert uri.port() is None + + def test_no_fragment_returns_empty(self): + uri = Uri.of("https://example.com/") + assert uri.fragment() == "" + + def test_no_query_returns_empty_dict(self): + uri = Uri.of("https://example.com/") + assert uri.query() == {} + + +class TestWithScheme: + def test_changes_scheme(self): + uri = Uri.of("http://example.com/").with_scheme("https") + assert uri.scheme() == "https" + assert uri.get().startswith("https://") + + def test_original_unchanged(self): + original = Uri.of("http://example.com/") + original.with_scheme("https") + assert original.scheme() == "http" + + +class TestWithHost: + def test_replaces_host(self): + uri = Uri.of("https://example.com/path").with_host("other.org") + assert uri.host() == "other.org" + assert "other.org" in uri.get() + + def test_preserves_path(self): + uri = Uri.of("https://example.com/api/v1").with_host("api.other.com") + assert uri.path() == "/api/v1" + + def test_preserves_query(self): + uri = Uri.of("https://example.com/?q=1").with_host("other.com") + assert uri.query_param("q") == "1" + + def test_preserves_port(self): + uri = Uri.of("https://example.com:9000/").with_host("other.com") + assert uri.port() == 9000 + + +class TestWithPort: + def test_sets_port(self): + uri = Uri.of("https://example.com/").with_port(8080) + assert uri.port() == 8080 + assert ":8080" in uri.get() + + def test_clears_port_with_none(self): + uri = Uri.of("https://example.com:9000/").with_port(None) + assert uri.port() is None + assert ":9000" not in uri.get() + + def test_preserves_host(self): + uri = Uri.of("https://example.com/").with_port(3000) + assert uri.host() == "example.com" + + +class TestWithPath: + def test_replaces_path(self): + uri = Uri.of("https://example.com/old").with_path("/new") + assert uri.path() == "/new" + + def test_preserves_query(self): + uri = Uri.of("https://example.com/old?x=1").with_path("/new") + assert uri.query_param("x") == "1" + assert uri.path() == "/new" + + +class TestAppendPath: + def test_appends_segment(self): + uri = Uri.of("https://example.com/api").append_path("users") + assert uri.path() == "/api/users" + + def test_handles_trailing_slash_on_base(self): + uri = Uri.of("https://example.com/api/").append_path("users") + assert uri.path() == "/api/users" + + def test_handles_leading_slash_on_segment(self): + uri = Uri.of("https://example.com/api").append_path("/users") + assert uri.path() == "/api/users" + + def test_chained_appends(self): + uri = ( + Uri.of("https://example.com/api") + .append_path("v1") + .append_path("users") + ) + assert uri.path() == "/api/v1/users" + + +class TestWithQuery: + def test_replaces_all_params(self): + uri = Uri.of("https://example.com/?a=1&b=2").with_query({"c": "3"}) + q = uri.query() + assert "a" not in q + assert "b" not in q + assert q["c"] == ["3"] + + def test_accepts_int_values(self): + uri = Uri.of("https://example.com/").with_query({"page": 5}) + assert uri.query_param("page") == "5" + + def test_accepts_list_values(self): + uri = Uri.of("https://example.com/").with_query({"ids": [1, 2, 3]}) + assert uri.query()["ids"] == ["1", "2", "3"] + + +class TestAddQuery: + def test_adds_new_param(self): + uri = Uri.of("https://example.com/?a=1").add_query("b", "2") + assert uri.query_param("a") == "1" + assert uri.query_param("b") == "2" + + def test_replaces_existing_param(self): + uri = Uri.of("https://example.com/?page=1").add_query("page", 2) + assert uri.query_param("page") == "2" + + def test_int_value_converted(self): + uri = Uri.of("https://example.com/").add_query("n", 42) + assert uri.query_param("n") == "42" + + +class TestRemoveQuery: + def test_removes_existing_param(self): + uri = Uri.of("https://example.com/?a=1&b=2").remove_query("a") + assert uri.query_param("a") is None + assert uri.query_param("b") == "2" + + def test_noop_on_missing_key(self): + uri = Uri.of("https://example.com/?a=1").remove_query("z") + assert uri.query_param("a") == "1" + + +class TestWithFragment: + def test_sets_fragment(self): + uri = Uri.of("https://example.com/page").with_fragment("section-2") + assert uri.fragment() == "section-2" + assert "#section-2" in uri.get() + + def test_replaces_existing_fragment(self): + uri = Uri.of("https://example.com/#old").with_fragment("new") + assert uri.fragment() == "new" + + +class TestWithoutFragment: + def test_clears_fragment(self): + uri = Uri.of("https://example.com/#top").without_fragment() + assert uri.fragment() == "" + assert "#" not in uri.get() + + +class TestWithoutQuery: + def test_clears_all_params(self): + uri = Uri.of("https://example.com/?a=1&b=2").without_query() + assert uri.query() == {} + assert "?" not in uri.get() + + +class TestFluencyChaining: + def test_full_chain(self): + url = ( + Uri.of("http://old.example.com/api?page=1#old") + .with_scheme("https") + .with_host("new.example.com") + .with_path("/v2/users") + .add_query("page", 2) + .add_query("limit", 50) + .remove_query("page") + .with_fragment("results") + .get() + ) + assert url == "https://new.example.com/v2/users?limit=50#results" + + def test_immutability_across_chain(self): + original = Uri.of("https://example.com/path") + step1 = original.with_scheme("http") + step2 = step1.add_query("x", "1") + + assert original.scheme() == "https" + assert original.query() == {} + assert step1.scheme() == "http" + assert step1.query() == {} + assert step2.scheme() == "http" + assert step2.query_param("x") == "1" + + +class TestEquality: + def test_equal_to_same_url_string(self): + uri = Uri.of("https://example.com/") + assert uri == "https://example.com/" + + def test_equal_to_uriable_with_same_url(self): + a = Uri.of("https://example.com/path") + b = Uri.of("https://example.com/path") + assert a == b + + def test_not_equal_to_different_url(self): + a = Uri.of("https://example.com/a") + b = Uri.of("https://example.com/b") + assert a != b + + +class TestRepr: + def test_repr_includes_url(self): + uri = Uri.of("https://example.com/") + assert "https://example.com/" in repr(uri) From 7e05a714004473be61ad854415e5d4d7e7c892c5 Mon Sep 17 00:00:00 2001 From: Bedram Tamang Date: Sat, 13 Jun 2026 09:44:37 -0700 Subject: [PATCH 2/2] fix: ruff lint and format in support/uri.py Remove unused `pytest` import and reformat chained method call in tests/utils/test_uri.py to satisfy ruff lint (F401) and ruff format checks. Co-Authored-By: Claude Sonnet 4.6 --- fastapi_startkit/tests/utils/test_uri.py | 8 +------- 1 file changed, 1 insertion(+), 7 deletions(-) diff --git a/fastapi_startkit/tests/utils/test_uri.py b/fastapi_startkit/tests/utils/test_uri.py index a3925b0b..e8396e7a 100644 --- a/fastapi_startkit/tests/utils/test_uri.py +++ b/fastapi_startkit/tests/utils/test_uri.py @@ -1,7 +1,5 @@ """Tests for the Uri fluent parser (task #178).""" -import pytest - from fastapi_startkit.support.uri import Uri, Uriable @@ -141,11 +139,7 @@ def test_handles_leading_slash_on_segment(self): assert uri.path() == "/api/users" def test_chained_appends(self): - uri = ( - Uri.of("https://example.com/api") - .append_path("v1") - .append_path("users") - ) + uri = Uri.of("https://example.com/api").append_path("v1").append_path("users") assert uri.path() == "/api/v1/users"