diff --git a/fastapi_startkit/src/fastapi_startkit/jsonapi/__init__.py b/fastapi_startkit/src/fastapi_startkit/jsonapi/__init__.py index d958cfb2..3d4c877a 100644 --- a/fastapi_startkit/src/fastapi_startkit/jsonapi/__init__.py +++ b/fastapi_startkit/src/fastapi_startkit/jsonapi/__init__.py @@ -1,15 +1,15 @@ """fastapi_startkit.jsonapi — JSON:API specification helpers.""" from .response import ( - JsonAPIListResponse, - JsonAPIResponse, + JsonResource, + ResourceCollection, parse_fields, parse_include, ) __all__ = [ - "JsonAPIResponse", - "JsonAPIListResponse", + "JsonResource", + "ResourceCollection", "parse_include", "parse_fields", ] diff --git a/fastapi_startkit/src/fastapi_startkit/jsonapi/response.py b/fastapi_startkit/src/fastapi_startkit/jsonapi/response.py index be32c81c..4e2d8388 100644 --- a/fastapi_startkit/src/fastapi_startkit/jsonapi/response.py +++ b/fastapi_startkit/src/fastapi_startkit/jsonapi/response.py @@ -4,48 +4,57 @@ Quick-start:: - from fastapi_startkit.jsonapi import JsonAPIResponse, JsonAPIListResponse - - class UserResource(JsonAPIResponse): - type = "users" - attributes = ["name", "email"] - - def __init__(self, user): - self.id = user.id - self.name = user.name - self.email = user.email - - # ── Single resource endpoint ──────────────────────────────────────── - @app.get("/api/users/{id}") - async def get_user(id: int): - user = await User.find_or_fail(id) - return UserResource(user) # ← return directly; FastAPI - # parses ?include= and - # fields[users]= for you - - # ── Collection endpoint ───────────────────────────────────────────── - @app.get("/api/users") - async def list_users(): - users = await User.all() - return JsonAPIListResponse([UserResource(u) for u in users]) - -Both classes implement the ASGI ``__call__`` protocol, so they can be -returned from FastAPI endpoints without any extra wiring. ``?include=`` and -``fields[*]=`` query params are parsed automatically from the live request. - -You can also call ``serialize()`` manually when you need the dict:: - - doc = UserResource(user).serialize( - include=["posts"], - fields={"users": ["name"], "posts": ["title"]}, - ) + from fastapi_startkit.jsonapi import JsonResource + + class PostResource(JsonResource["Post"]): + pass # auto-type="posts", auto-attributes from Post.serialize() + + class UserResource(JsonResource["User"]): + hidden = ["password", "remember_token"] + + class ArticleResource(JsonResource["Article"]): + def with_(self): + return {"meta": {"version": "1.0"}} + + # Single resource — returned directly; ?include= and ?fields[*]= are + # parsed from the live request automatically. + @app.get("/api/posts/{id}") + async def get_post(id: int): + post = await Post.find_or_fail(id) + return PostResource(post) + + # Restrict fields and sideload relationships with the fluent chain API: + @app.get("/api/posts/{id}") + async def get_post(id: int): + post = await Post.find_or_fail(id) + return PostResource(post).include("author").fields("title", "users.name") + + # Collection (plain list or paginator) + @app.get("/api/posts") + async def list_posts(): + posts = await Post.all() + return PostResource.collection(posts) + + # Paginated collection + @app.get("/api/posts") + async def list_posts_paginated(page: int = 1): + posts = await Post.paginate(15, page) + return PostResource.collection(posts) + + # Manual serialization when you need the dict: + doc = PostResource(post).include("author").fields("title", "users.name").serialize() """ from __future__ import annotations -from typing import Any +import inspect +from typing import Any, Generic, TypeVar from urllib.parse import unquote_plus +import inflection + +T = TypeVar("T") + # --------------------------------------------------------------------------- # Query-param helpers @@ -60,16 +69,6 @@ def parse_include(param: str | None) -> list[str]: :param param: raw query-string value or ``None``. :returns: list of relationship names to sideload. - - Example FastAPI usage:: - - from fastapi import Query - from fastapi_startkit.jsonapi import parse_include - - @app.get("/api/posts/{id}") - async def get_post(id: int, include: str | None = Query(None)): - post = await Post.find(id) - return PostResource(post).serialize(include=parse_include(include)) """ if not param: return [] @@ -85,19 +84,6 @@ def parse_fields(raw_query: dict[str, str]) -> dict[str, list[str]]: :param raw_query: flat ``{key: value}`` dict of ALL query parameters (e.g. ``dict(request.query_params)`` in FastAPI). :returns: ``{resource_type: [field, ...]}`` for every ``fields[*]`` key. - - Example FastAPI usage:: - - from fastapi import Request - from fastapi_startkit.jsonapi import parse_fields - - @app.get("/api/posts/{id}") - async def get_post(id: int, request: Request): - fields = parse_fields(dict(request.query_params)) - # GET ?fields[posts]=title,body&fields[users]=name - # → {"posts": ["title", "body"], "users": ["name"]} - post = await Post.find(id) - return PostResource(post).serialize(fields=fields) """ result: dict[str, list[str]] = {} for key, value in raw_query.items(): @@ -112,29 +98,18 @@ async def get_post(id: int, request: Request): # ASGI mixin — makes resources directly returnable from FastAPI endpoints # --------------------------------------------------------------------------- -# FastAPI checks `isinstance(response, starlette.responses.Response)` on every -# endpoint return value. If True, it calls `await response(scope, receive, send)` -# directly. If False, it JSON-serialises the raw object __dict__. -# -# We therefore make _FastAPICallable a real Response subclass when starlette is -# available, and fall back to plain `object` otherwise (serialize() still works, -# the user just can't return the resource directly from FastAPI). - try: from starlette.responses import Response as _StarletteResponse class _FastAPICallable(_StarletteResponse): """Starlette Response subclass that lazily serializes on ``__call__``. - Subclasses (UserResource, PostResource, …) have their own ``__init__`` - and never call ``super().__init__()``, so we must NOT rely on Starlette's - ``Response.__init__`` having run. We set the attributes FastAPI reads - before calling ``__call__`` as class-level defaults, and we completely - override ``__call__`` to handle ASGI ourselves. + If chain state has been set via ``.include()`` / ``.fields()`` before + the resource is returned from an endpoint, that state is used. + Otherwise the query string is parsed from the ASGI scope and passed + to ``serialize()`` automatically. """ - # FastAPI reads `response.background` before calling `__call__`. - # Setting it at class level ensures it's always present. background = None status_code = 200 media_type = "application/vnd.api+json" @@ -142,20 +117,23 @@ class _FastAPICallable(_StarletteResponse): async def __call__(self, scope: Any, receive: Any, send: Any) -> None: import json as _json - # Parse query string from the ASGI scope. - raw_qs: str = (scope.get("query_string") or b"").decode() - qp: dict[str, str] = {} - if raw_qs: - for pair in raw_qs.split("&"): - if "=" in pair: - k, _, v = pair.partition("=") - qp[unquote_plus(k)] = unquote_plus(v) - - include = parse_include(qp.get("include")) - fields = parse_fields(qp) + # If chain methods were called before returning this resource, + # _chain_state_set is True and we use that state directly. + # Otherwise fall back to parsing ?include= and ?fields[*]= from the URL. + if not getattr(self, "_chain_state_set", False): + raw_qs: str = (scope.get("query_string") or b"").decode() + qp: dict[str, str] = {} + if raw_qs: + for pair in raw_qs.split("&"): + if "=" in pair: + k, _, v = pair.partition("=") + qp[unquote_plus(k)] = unquote_plus(v) + # Temporarily set chain state from query string for this render. + self._chain_include = parse_include(qp.get("include")) # type: ignore[attr-defined] + self._chain_fields = parse_fields(qp) # type: ignore[attr-defined] body = _json.dumps( - self.serialize(include=include, fields=fields) # type: ignore[attr-defined] + self.serialize() # type: ignore[attr-defined] ).encode("utf-8") await send( @@ -170,50 +148,138 @@ async def __call__(self, scope: Any, receive: Any, send: Any) -> None: ) await send({"type": "http.response.body", "body": body}) -except ImportError: # starlette / fastapi not installed +except ImportError: class _FastAPICallable: # type: ignore[no-redef] - """No-op when starlette is not installed. - - ``serialize()`` still works; returning the resource directly from a - FastAPI endpoint will not until ``fastapi-startkit[fastapi]`` is installed. - """ + """No-op when starlette is not installed.""" # --------------------------------------------------------------------------- -# Core classes +# JsonResource — generic base class # --------------------------------------------------------------------------- -class JsonAPIResponse(_FastAPICallable): - """Base class for a single JSON:API resource. +class JsonResource(Generic[T], _FastAPICallable): + """Generic base class for a single JSON:API resource. + + Pass the model directly:: + + class PostResource(JsonResource[Post]): + pass # auto-type="posts", auto-attributes from Post.serialize() - Subclasses must define: + class UserResource(JsonResource[User]): + hidden = ["password"] # strip sensitive fields - * ``type: str`` – resource type (e.g. ``"posts"``) - * ``id: int | str`` – resource identifier (set in ``__init__``) + class ArticleResource(JsonResource[Article]): + def with_(self): + return {"meta": {"version": "1.0"}} - Optionally declare: + Fluent chain API:: - * ``attributes: list[str]`` - Field names to expose in ``data.attributes``. The default - ``to_attributes()`` reads the matching instance attributes. + # Sideload relationships + return PostResource(post).include("author", "comments") - * ``relationships: dict[str, JsonAPIResponse]`` - Related resources (populated in ``to_relationships()``). + # Restrict attributes — plain names apply to this resource's type; + # dotted names ("type.field") apply to a related resource's type. + return PostResource(post).fields("title", "created_at", "users.name") - All four hooks are overridable: ``to_attributes()``, - ``to_relationships()``, ``to_links()``, ``to_meta()``. + # Chain include + fields + return PostResource(post).include("author").fields("title", "users.name") - Instances can be returned **directly** from FastAPI endpoints — the - ASGI ``__call__`` in :class:`_FastAPICallable` handles query-param - parsing and serialization automatically. + # Manual serialization + doc = PostResource(post).include("author").fields("title", "users.name").serialize() + + When returned directly from a FastAPI endpoint with no chain methods called, + ``?include=`` and ``?fields[*]=`` query params are parsed automatically + from the live request. + + Class-level attributes + ---------------------- + type : str + Resource type string. Auto-derived from the class name when not set + (``AgentResource`` -> ``"agents"``). + id : int | str + Resource identifier. Auto-set from ``model.id`` in ``__init__``. + hidden : list[str] + Field names to exclude from ``to_attributes()`` when auto-serializing + via ``model.serialize()``. + + Class methods + ------------- + collection(items) + Wrap a plain list **or** a ``LengthAwarePaginator`` / ``SimplePaginator`` + in a :class:`ResourceCollection`. Pagination meta is included + automatically. + + Overridable hooks + ----------------- + to_attributes() -- ``{name: value}`` dict of resource attributes + to_relationships() -- ``{name: JsonResource}`` related resources + to_links() -- top-level links dict + to_meta() -- top-level meta dict + with_() -- extra top-level envelope keys merged into the document """ type: str = "" id: int | str = "" - attributes: list[str] = [] - relationships: dict[str, "JsonAPIResponse"] = {} + hidden: list[str] = [] + + def __init_subclass__(cls, **kwargs: Any) -> None: + super().__init_subclass__(**kwargs) + if "type" not in cls.__dict__: + name = cls.__name__.removesuffix("Resource") + if name: + cls.type = inflection.tableize(name) + + def __init__(self, model: T) -> None: + self.model = model + self.id = getattr(model, "id", "") + self._chain_include: list[str] = [] + self._chain_fields: dict[str, list[str]] = {} + self._chain_state_set: bool = False + + # ------------------------------------------------------------------ + # Fluent chain API + # ------------------------------------------------------------------ + + def include(self, *relationships: str) -> "JsonResource[T]": + """Sideload one or more relationships into ``included[]``. + + Dot-notation is supported for nested relationships:: + + PostResource(post).include("author", "author.company") + + :param relationships: relationship names to sideload. + :returns: ``self`` for chaining. + """ + self._chain_include = list(relationships) + self._chain_state_set = True + return self + + def fields(self, *field_specs: str) -> "JsonResource[T]": + """Restrict which attributes are returned (JSON:API sparse fieldsets). + + Each argument is either a plain field name or a ``"type.field"`` spec: + + * Plain name (``"title"``) — restricts this resource's own attributes. + * Dotted name (``"users.name"``) — restricts the named related type. + + Example:: + + PostResource(post).fields("title", "created_at", "users.name") + # → {"posts": ["title", "created_at"], "users": ["name"]} + + :param field_specs: field names, optionally prefixed with ``type.``. + :returns: ``self`` for chaining. + """ + for spec in field_specs: + if "." in spec: + rel_type, _, field = spec.partition(".") + self._chain_fields.setdefault(rel_type, []).append(field) + else: + self._chain_fields.setdefault(self.type, []).append(spec) + self._chain_state_set = True + return self # ------------------------------------------------------------------ # Overridable hooks @@ -222,26 +288,53 @@ class JsonAPIResponse(_FastAPICallable): def to_attributes(self) -> dict | None: """Return a ``{name: value}`` dict of resource attributes. - The default implementation builds the dict from ``self.attributes`` - by reading matching instance attributes (``None`` when absent). - Returns ``None`` when ``self.attributes`` is empty. + Calls ``self.model.serialize()`` and strips any names listed in + ``self.hidden``. All other fields are included as-is. + Returns ``None`` when the model has no ``serialize()`` method or + when the resulting dict is empty. """ - if not self.attributes: - return None - return {attr: getattr(self, attr, None) for attr in self.attributes} + model = getattr(self, "model", None) + if model is not None and hasattr(model, "serialize"): + data = model.serialize() + if isinstance(data, dict): + blacklist = set(self.__class__.hidden) + filtered = {k: v for k, v in data.items() if k not in blacklist} + return filtered or None + return None + + def to_relationships(self) -> dict | None: + """Return a relationship mapping, or ``None``. + + Each value can be any of three forms: + + * **Resource class** → always a **single** resource. + The framework reads ``self.model.`` and wraps it:: + + def to_relationships(self): + return {"author": UserResource} + # → UserResource(self.model.author) + + If the attribute is ``None`` or absent the relationship is omitted. + + * **Lambda / function** → for **collections** and custom logic. + Called with no arguments; use ``ResourceClass.collection()`` inside:: + + def to_relationships(self): + return { + "comments": lambda: CommentResource.collection( + self.model.comments + ), + } - def to_relationships(self) -> dict[str, "JsonAPIResponse"] | None: - """Return a ``{name: JsonAPIResponse}`` dict of related resources. + * **``JsonResource`` / ``ResourceCollection`` instance** — used as-is + when you need full control:: - The default implementation returns the class-level - ``relationships`` dict only when its values are already - ``JsonAPIResponse`` instances. Override this to build the dict - dynamically from an ORM object. + def to_relationships(self): + return {"author": UserResource(self.model.author)} + + Returns ``None`` (no relationships) by default. """ - rels = self.__class__.relationships - if not rels: - return None - return rels if all(isinstance(v, JsonAPIResponse) for v in rels.values()) else None + return None def to_links(self) -> dict | None: """Return a ``{name: url}`` dict of links, or ``None``.""" @@ -251,17 +344,71 @@ def to_meta(self) -> dict | None: """Return a ``{name: value}`` meta dict, or ``None``.""" return None + def with_(self) -> dict: + """Return extra keys to merge into the top-level JSON:API envelope. + + Keys from ``with_()`` are shallow-merged last, so they take + precedence over ``to_links()`` / ``to_meta()``:: + + class ArticleResource(JsonResource[Article]): + def with_(self): + return {"meta": {"version": "1.0"}} + + :returns: dict of extra top-level envelope keys (default: ``{}``). + """ + return {} + # ------------------------------------------------------------------ - # Internal serialization helpers + # Internal helpers # ------------------------------------------------------------------ - def _build_data(self, fields: dict[str, list[str]] | None = None) -> dict: - """Build the ``data`` member of the JSON:API document. + def _resolve_rel(self, key: str, value: Any) -> "JsonResource | ResourceCollection | None": + """Resolve one relationship value to a concrete resource / collection. + + Two intended forms: + + * **Resource class** → always a **single** resource. + The framework reads ``self.model.`` and wraps it with the class:: + + return {"author": UserResource} + # → UserResource(self.model.author) + + * **Lambda / function** → for **collections** and any custom logic. + Called with no arguments; use ``ResourceClass.collection()`` inside:: + + return { + "comments": lambda: CommentResource.collection(self.model.comments), + } - :param fields: sparse-fieldset map from :func:`parse_fields`. - When present, only the listed fields are included in - ``data.attributes`` for each resource type. + Also accepted for convenience: + + * **``JsonResource`` / ``ResourceCollection`` instance** — used as-is. """ + # --- class reference → single resource --- + if isinstance(value, type): + related = getattr(self.model, key, None) if hasattr(self, "model") else None + return value(related) if related is not None else None + + # --- lambda / plain function --- + if inspect.isfunction(value) or inspect.ismethod(value): + return value() + + # --- already a JsonResource or ResourceCollection instance --- + return value + + def _resolved_relationships(self) -> "dict[str, JsonResource | ResourceCollection]": + """Return ``to_relationships()`` with all values resolved to instances.""" + raw = self.to_relationships() + if not raw: + return {} + resolved: dict[str, Any] = {} + for key, val in raw.items(): + result = self._resolve_rel(key, val) + if result is not None: + resolved[key] = result + return resolved + + def _build_data(self, fields: dict[str, list[str]] | None = None) -> dict: data: dict[str, Any] = { "type": self.type, "id": str(self.id), @@ -274,11 +421,15 @@ def _build_data(self, fields: dict[str, list[str]] | None = None) -> dict: attrs = {k: v for k, v in attrs.items() if k in allowed} data["attributes"] = attrs - rel_objs = self.to_relationships() + rel_objs = self._resolved_relationships() if rel_objs: - data["relationships"] = { - name: {"data": {"type": resource.type, "id": str(resource.id)}} for name, resource in rel_objs.items() - } + rels: dict[str, Any] = {} + for name, resource in rel_objs.items(): + if isinstance(resource, ResourceCollection): + rels[name] = {"data": [{"type": item.type, "id": str(item.id)} for item in resource._items]} + else: + rels[name] = {"data": {"type": resource.type, "id": str(resource.id)}} + data["relationships"] = rels return data @@ -288,81 +439,49 @@ def _collect_included( seen: set[str] | None = None, fields: dict[str, list[str]] | None = None, ) -> list[dict]: - """Recursively sideload related resources. - - :param include: relationship names to sideload. - :param seen: de-duplication set of ``"type:id"`` keys. - :param fields: sparse-fieldset filter. - """ if seen is None: seen = set() included: list[dict] = [] - rel_objs = self.to_relationships() or {} + rel_objs = self._resolved_relationships() for name, resource in rel_objs.items(): if name not in include: continue - - key = f"{resource.type}:{resource.id}" - if key in seen: - continue - seen.add(key) - included.append(resource._build_data(fields=fields)) - - # Recurse for nested dot-notation includes (e.g. "author.company"). + # Handle collection relationships + items = resource._items if isinstance(resource, ResourceCollection) else [resource] nested = [part[len(name) + 1 :] for part in include if part.startswith(f"{name}.")] - if nested: - included.extend(resource._collect_included(nested, seen, fields=fields)) + for item in items: + key = f"{item.type}:{item.id}" + if key in seen: + continue + seen.add(key) + included.append(item._build_data(fields=fields)) + if nested: + included.extend(item._collect_included(nested, seen, fields=fields)) return included # ------------------------------------------------------------------ - # Public API + # Serialization # ------------------------------------------------------------------ - def serialize( - self, - include: list[str] | None = None, - fields: dict[str, list[str]] | None = None, - ) -> dict: + def serialize(self) -> dict: """Serialize this resource into a JSON:API document dict. - :param include: relationship names to sideload into ``included[]``. - Use comma-separated names via :func:`parse_include`, - or pass a plain list. Dot notation is supported for - nested relationships (``"author.company"``). - :param fields: sparse-fieldset map produced by :func:`parse_fields`. - Only the listed attribute names are included for each - resource type: - ``{"posts": ["title"], "users": ["name"]}``. - - :returns: A dict safe to return from any FastAPI endpoint or pass - to ``JSONResponse``. - - Typical FastAPI endpoint (manual):: - - @app.get("/api/posts/{id}") - async def get_post( - id: int, - request: Request, - include: str | None = Query(None), - ): - post = await Post.find(id) - return PostResource(post).serialize( - include=parse_include(include), - fields=parse_fields(dict(request.query_params)), - ) - - Or simply return the resource directly and let FastAPI handle it:: - - @app.get("/api/posts/{id}") - async def get_post(id: int): - post = await Post.find(id) - return PostResource(post) + Uses the include / fields state set via the chain API:: + + doc = PostResource(post).include("author").fields("posts", ["title"]).serialize() + + When returned directly from a FastAPI endpoint without calling chain + methods, ``?include=`` and ``?fields[*]=`` query params are parsed + from the live HTTP request automatically. + + :returns: A dict safe to pass to ``JSONResponse`` or return from + any FastAPI endpoint. """ - if include is None: - include = [] + include = self._chain_include + fields = self._chain_fields or None document: dict[str, Any] = {"data": self._build_data(fields=fields)} @@ -379,47 +498,152 @@ async def get_post(id: int): if meta is not None: document["meta"] = meta + extra = self.with_() + if extra: + document.update(extra) + return document + # ------------------------------------------------------------------ + # Collection factory + # ------------------------------------------------------------------ + + @classmethod + def collection(cls, items: Any) -> "ResourceCollection": + """Wrap *items* in a :class:`ResourceCollection`. + + *items* may be: + + * A plain ``list`` or iterable of model instances. + * A ``LengthAwarePaginator`` or ``SimplePaginator`` — pagination + meta is included automatically in the response envelope. + + Example:: + + return PostResource.collection(posts) + return PostResource.collection(posts).include("author") + return PostResource.collection(paginator).fields("posts", ["title"]) + """ + try: + from fastapi_startkit.masoniteorm.pagination.BasePaginator import BasePaginator + + if isinstance(items, BasePaginator): + resource_items = [cls(model) for model in items] + return ResourceCollection(resource_items, paginator=items, primary_type=cls.type) + except ImportError: + pass + + return ResourceCollection([cls(model) for model in items], primary_type=cls.type) + + +# --------------------------------------------------------------------------- +# ResourceCollection +# --------------------------------------------------------------------------- -class JsonAPIListResponse(_FastAPICallable): - """Wraps a list of :class:`JsonAPIResponse` instances. - ``data`` is a JSON array. Instances can be returned directly from - FastAPI endpoints — query params are parsed automatically. +class ResourceCollection(_FastAPICallable): + """Wraps a list of :class:`JsonResource` instances as a JSON:API collection. - Usage:: + Prefer creating instances via :meth:`JsonResource.collection`:: - @app.get("/api/posts") - async def list_posts(): - posts = await Post.all() - return JsonAPIListResponse([PostResource(p) for p in posts]) + return PostResource.collection(posts) + return PostResource.collection(posts).include("author") + return PostResource.collection(posts).fields("title", "users.name") + + Pagination meta is added automatically when the source is a paginator. + Override :meth:`to_meta` / :meth:`to_links` to add custom envelope data. """ - def __init__(self, items: list[JsonAPIResponse]) -> None: + def __init__( + self, + items: list[JsonResource], + paginator: Any = None, + primary_type: str = "", + ) -> None: self._items = items + self._paginator = paginator + self._primary_type = primary_type + self._chain_include: list[str] = [] + self._chain_fields: dict[str, list[str]] = {} + self._chain_state_set: bool = False + + # ------------------------------------------------------------------ + # Fluent chain API + # ------------------------------------------------------------------ + + def include(self, *relationships: str) -> "ResourceCollection": + """Sideload one or more relationships into ``included[]``. + + :param relationships: relationship names to sideload. + :returns: ``self`` for chaining. + """ + self._chain_include = list(relationships) + self._chain_state_set = True + return self + + def fields(self, *field_specs: str) -> "ResourceCollection": + """Restrict which attributes are returned (JSON:API sparse fieldsets). + + Same dot-notation as :meth:`JsonResource.fields` — plain names apply + to the primary resource type; ``"type.field"`` applies to a related type:: + + PostResource.collection(posts).fields("title", "created_at", "users.name") + + :param field_specs: field names, optionally prefixed with ``type.``. + :returns: ``self`` for chaining. + """ + for spec in field_specs: + if "." in spec: + rel_type, _, field = spec.partition(".") + self._chain_fields.setdefault(rel_type, []).append(field) + else: + self._chain_fields.setdefault(self._primary_type, []).append(spec) + self._chain_state_set = True + return self + + # ------------------------------------------------------------------ + # Overridable hooks + # ------------------------------------------------------------------ def to_links(self) -> dict | None: """Return top-level links, or ``None``.""" return None def to_meta(self) -> dict | None: - """Return top-level meta, or ``None``.""" - return None + """Return top-level meta dict, or ``None``. - def serialize( - self, - include: list[str] | None = None, - fields: dict[str, list[str]] | None = None, - ) -> dict: + Auto-populated from a paginator when present. + """ + if self._paginator is None: + return None + + paginator = self._paginator + meta: dict[str, Any] = {} + for attr in ( + "total", + "count", + "per_page", + "current_page", + "last_page", + "next_page", + "previous_page", + ): + if hasattr(paginator, attr): + meta[attr] = getattr(paginator, attr) + + return meta if meta else None + + # ------------------------------------------------------------------ + # Serialization + # ------------------------------------------------------------------ + + def serialize(self) -> dict: """Serialize the collection into a JSON:API document dict. - :param include: relationship names to sideload (same semantics as - :meth:`JsonAPIResponse.serialize`). - :param fields: sparse-fieldset map from :func:`parse_fields`. + Uses the include / fields state set via the chain API. """ - if include is None: - include = [] + include = self._chain_include + fields = self._chain_fields or None document: dict[str, Any] = { "data": [item._build_data(fields=fields) for item in self._items], diff --git a/fastapi_startkit/tests/jsonapi/test_include.py b/fastapi_startkit/tests/jsonapi/test_include.py index 5356b974..88a520fe 100644 --- a/fastapi_startkit/tests/jsonapi/test_include.py +++ b/fastapi_startkit/tests/jsonapi/test_include.py @@ -1,74 +1,79 @@ -"""Tests for include= sideloading logic in JsonAPIResponse.""" +"""Tests for include sideloading via the .include() chain method.""" -from fastapi_startkit.jsonapi import JsonAPIResponse, JsonAPIListResponse +from fastapi_startkit.jsonapi import JsonResource, ResourceCollection # --------------------------------------------------------------------------- -# Fixture resources — three-level hierarchy: Post → Author → Company +# Fake models # --------------------------------------------------------------------------- -class CompanyResource(JsonAPIResponse): - type = "companies" - attributes = ["name"] - +class FakeCompany: def __init__(self, id_, name): self.id = id_ self.name = name - def to_attributes(self): - return {"name": self.name} - + def serialize(self): + return {"id": self.id, "name": self.name} -class AuthorResource(JsonAPIResponse): - type = "authors" - attributes = ["username"] +class FakeAuthor: def __init__(self, id_, username, company=None): self.id = id_ self.username = username - self._company = company - - def to_attributes(self): - return {"username": self.username} - - def to_relationships(self): - if self._company is None: - return None - return {"company": self._company} + self.company = company + def serialize(self): + return {"id": self.id, "username": self.username} -class CommentResource(JsonAPIResponse): - type = "comments" - attributes = ["body"] +class FakeComment: def __init__(self, id_, body): self.id = id_ self.body = body - def to_attributes(self): - return {"body": self.body} + def serialize(self): + return {"id": self.id, "body": self.body} -class PostResource(JsonAPIResponse): - type = "posts" - attributes = ["title"] - +class FakePost: def __init__(self, id_, title, author=None, comments=None): self.id = id_ self.title = title - self._author = author - self._comments = comments or [] + self.author = author + self.comments = comments or [] + + def serialize(self): + return {"id": self.id, "title": self.title} + - def to_attributes(self): - return {"title": self.title} +# --------------------------------------------------------------------------- +# Fixture resources — three-level hierarchy: Post -> Author -> Company +# --------------------------------------------------------------------------- + +class CompanyResource(JsonResource[FakeCompany]): + pass # type="companies", auto-attrs + + +class AuthorResource(JsonResource[FakeAuthor]): + def to_relationships(self): + if self.model.company is None: + return None + return {"company": CompanyResource(self.model.company)} + + +class CommentResource(JsonResource[FakeComment]): + pass # type="comments", auto-attrs + + +class PostResource(JsonResource[FakePost]): def to_relationships(self): rels = {} - if self._author is not None: - rels["author"] = self._author - for i, comment in enumerate(self._comments): - rels[f"comment_{i}"] = comment + if self.model.author is not None: + rels["author"] = AuthorResource(self.model.author) + for i, comment in enumerate(self.model.comments): + rels[f"comment_{i}"] = CommentResource(comment) return rels or None @@ -79,48 +84,46 @@ def to_relationships(self): class TestIncludeSingleRelationship: def test_include_author_present(self): - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) - doc = post.serialize(include=["author"]) + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) + doc = post.include("author").serialize() assert "included" in doc assert any(inc["type"] == "authors" and inc["id"] == "1" for inc in doc["included"]) def test_include_uses_resource_attributes(self): - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) - doc = post.serialize(include=["author"]) + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) + doc = post.include("author").serialize() inc = next(i for i in doc["included"] if i["type"] == "authors") assert inc["attributes"]["username"] == "alice" def test_include_non_existent_relationship_no_included(self): - post = PostResource(10, "My Post") - doc = post.serialize(include=["author"]) + post = PostResource(FakePost(10, "My Post")) + doc = post.include("author").serialize() assert "included" not in doc def test_include_none_equivalent_to_empty(self): - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) - doc_none = post.serialize(include=None) - doc_empty = post.serialize(include=[]) - assert "included" not in doc_none - assert "included" not in doc_empty + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) + doc = post.serialize() + assert "included" not in doc class TestIncludeMultipleRelationships: def test_include_multiple_keys(self): - author = AuthorResource(1, "alice") - comment = CommentResource(100, "Great post!") - post = PostResource(10, "My Post", author=author, comments=[comment]) - doc = post.serialize(include=["author", "comment_0"]) + author = FakeAuthor(1, "alice") + comment = FakeComment(100, "Great post!") + post = PostResource(FakePost(10, "My Post", author=author, comments=[comment])) + doc = post.include("author", "comment_0").serialize() types = {inc["type"] for inc in doc["included"]} assert "authors" in types assert "comments" in types def test_include_only_requested_keys(self): - author = AuthorResource(1, "alice") - comment = CommentResource(100, "Great post!") - post = PostResource(10, "My Post", author=author, comments=[comment]) - doc = post.serialize(include=["author"]) + author = FakeAuthor(1, "alice") + comment = FakeComment(100, "Great post!") + post = PostResource(FakePost(10, "My Post", author=author, comments=[comment])) + doc = post.include("author").serialize() types = {inc["type"] for inc in doc["included"]} assert "authors" in types assert "comments" not in types @@ -128,29 +131,29 @@ def test_include_only_requested_keys(self): class TestIncludeDeduplication: def test_same_resource_not_duplicated(self): - author = AuthorResource(1, "alice") + author = FakeAuthor(1, "alice") # Two posts sharing the same author object. posts = [ - PostResource(1, "P1", author=author), - PostResource(2, "P2", author=author), + PostResource(FakePost(1, "P1", author=author)), + PostResource(FakePost(2, "P2", author=author)), ] - doc = JsonAPIListResponse(posts).serialize(include=["author"]) + doc = ResourceCollection(posts).include("author").serialize() author_entries = [i for i in doc["included"] if i["type"] == "authors"] assert len(author_entries) == 1 class TestIncludeRelationshipDataInDocument: def test_relationships_key_in_data_still_present(self): - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) - doc = post.serialize(include=["author"]) + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) + doc = post.include("author").serialize() assert "relationships" in doc["data"] assert "author" in doc["data"]["relationships"] def test_relationship_linkage_is_correct(self): - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) - doc = post.serialize(include=["author"]) + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) + doc = post.include("author").serialize() linkage = doc["data"]["relationships"]["author"]["data"] assert linkage == {"type": "authors", "id": "1"} @@ -159,26 +162,26 @@ class TestIncludeFromQueryParam: """Simulate how FastAPI would pass the include= query parameter.""" def test_include_from_comma_split(self): - """Demonstrate parsing include= query string and passing to serialize().""" - author = AuthorResource(1, "alice") - post = PostResource(10, "My Post", author=author) + """Demonstrate parsing include= query string and passing to .include().""" + author = FakeAuthor(1, "alice") + post = PostResource(FakePost(10, "My Post", author=author)) # FastAPI endpoint would do: include_param.split(",") include_param = "author" include_list = [x.strip() for x in include_param.split(",")] - doc = post.serialize(include=include_list) + doc = post.include(*include_list).serialize() assert "included" in doc def test_include_multiple_from_comma_split(self): - author = AuthorResource(1, "alice") - comment = CommentResource(100, "Nice") - post = PostResource(10, "My Post", author=author, comments=[comment]) + author = FakeAuthor(1, "alice") + comment = FakeComment(100, "Nice") + post = PostResource(FakePost(10, "My Post", author=author, comments=[comment])) include_param = "author, comment_0" include_list = [x.strip() for x in include_param.split(",")] - doc = post.serialize(include=include_list) + doc = post.include(*include_list).serialize() types = {inc["type"] for inc in doc["included"]} assert "authors" in types assert "comments" in types diff --git a/fastapi_startkit/tests/jsonapi/test_json_resource.py b/fastapi_startkit/tests/jsonapi/test_json_resource.py new file mode 100644 index 00000000..0dfb9e5d --- /dev/null +++ b/fastapi_startkit/tests/jsonapi/test_json_resource.py @@ -0,0 +1,515 @@ +"""Tests for the JsonResource[T] generic base class. + +Covers: +- Auto-type derivation from class name via inflection +- __init__(self, model) storing self.model and self.id +- Default to_attributes() using model.serialize() minus 'id' and hidden fields +- hidden class var blacklist +- with_() method merging into top-level envelope +- JsonResource.collection() with plain lists and paginators +""" + +from __future__ import annotations + +from fastapi_startkit.jsonapi import ( + JsonResource, + ResourceCollection, +) +from fastapi_startkit.masoniteorm.pagination.LengthAwarePaginator import ( + LengthAwarePaginator, +) +from fastapi_startkit.masoniteorm.pagination.SimplePaginator import SimplePaginator + + +# --------------------------------------------------------------------------- +# Fake model helpers +# --------------------------------------------------------------------------- + + +class _FakeModelCollection: + """Minimal list-like object with a .serialize() method (mimics ORM result).""" + + def __init__(self, rows): + self._rows = rows + + def serialize(self, *args, **kwargs): + return [r.__dict__ for r in self._rows] + + def __len__(self): + return len(self._rows) + + def __getitem__(self, item): + return self._rows[item] + + def __iter__(self): + return iter(self._rows) + + +class FakeModel: + """Minimal model with id and serialize().""" + + def __init__(self, id_, **fields): + self.id = id_ + for k, v in fields.items(): + setattr(self, k, v) + self._fields = {"id": id_, **fields} + + def serialize(self): + return dict(self._fields) + + +class FakeModelNoSerialize: + """Model without a serialize() method.""" + + def __init__(self, id_, name): + self.id = id_ + self.name = name + + +# --------------------------------------------------------------------------- +# Concrete resource fixtures +# --------------------------------------------------------------------------- + + +class PostResource(JsonResource[FakeModel]): + """Auto-type = "posts", auto-attributes from model.serialize().""" + + pass + + +class UserResource(JsonResource[FakeModel]): + """hidden fields exclude 'password'.""" + + hidden = ["password"] + + +class ArticleResource(JsonResource[FakeModel]): + """Injects extra top-level envelope keys via with_().""" + + def with_(self): + return {"meta": {"version": "1.0"}} + + +class AgentResource(JsonResource[FakeModel]): + """Tests multi-word class name -> tableized type.""" + + pass + + +class UserProfileResource(JsonResource[FakeModel]): + """Multi-word: UserProfile -> user_profiles.""" + + pass + + +# --------------------------------------------------------------------------- +# 1. Auto-type derivation +# --------------------------------------------------------------------------- + + +class TestAutoType: + def test_single_word_resource(self): + model = FakeModel(1, title="Hello") + resource = PostResource(model) + assert resource.type == "posts" + + def test_single_word_user(self): + model = FakeModel(1, name="Alice") + resource = UserResource(model) + assert resource.type == "users" + + def test_multi_word_type(self): + model = FakeModel(1, name="Alice") + resource = UserProfileResource(model) + assert resource.type == "user_profiles" + + def test_agent_type(self): + model = FakeModel(1, name="Bot") + resource = AgentResource(model) + assert resource.type == "agents" + + def test_explicit_type_overrides_auto(self): + class CustomResource(JsonResource[FakeModel]): + type = "my_custom_type" + + model = FakeModel(1) + resource = CustomResource(model) + assert resource.type == "my_custom_type" + + def test_type_not_set_on_base_class(self): + # JsonResource itself has type="" — auto-type only fires for subclasses + assert JsonResource.type == "" + + +# --------------------------------------------------------------------------- +# 2. __init__ stores model and id +# --------------------------------------------------------------------------- + + +class TestInit: + def test_model_stored(self): + model = FakeModel(42, title="Test") + resource = PostResource(model) + assert resource.model is model + + def test_id_from_model(self): + model = FakeModel(99, title="Test") + resource = PostResource(model) + assert resource.id == 99 + + def test_id_defaults_to_empty_string_when_no_id(self): + class NoIdModel: + def serialize(self): + return {"name": "x"} + + class ThingResource(JsonResource[NoIdModel]): + pass + + m = NoIdModel() + r = ThingResource(m) + assert r.id == "" + + +# --------------------------------------------------------------------------- +# 3. Default to_attributes() auto-serialization +# --------------------------------------------------------------------------- + + +class TestToAttributesAutoSerialize: + def test_attributes_from_serialize(self): + model = FakeModel(1, title="Hello", body="World") + resource = PostResource(model) + attrs = resource.to_attributes() + assert attrs == {"id": 1, "title": "Hello", "body": "World"} + + def test_id_present_in_attributes(self): + model = FakeModel(1, title="Hello") + resource = PostResource(model) + attrs = resource.to_attributes() + assert "id" in attrs + assert attrs["id"] == 1 + + def test_returns_none_when_model_has_no_serialize(self): + class BareResource(JsonResource[FakeModelNoSerialize]): + pass + + model = FakeModelNoSerialize(1, "Alice") + resource = BareResource(model) + assert resource.to_attributes() is None + + def test_returns_none_when_serialize_returns_empty(self): + class EmptyModel: + id = 1 + + def serialize(self): + return {} + + class EmptyResource(JsonResource[EmptyModel]): + pass + + resource = EmptyResource(EmptyModel()) + assert resource.to_attributes() is None + + def test_in_serialized_document(self): + model = FakeModel(1, title="Hello", body="World") + doc = PostResource(model).serialize() + assert doc["data"]["attributes"] == {"id": 1, "title": "Hello", "body": "World"} + + +# --------------------------------------------------------------------------- +# 4. hidden class var +# --------------------------------------------------------------------------- + + +class TestHidden: + def test_hidden_field_excluded(self): + model = FakeModel(1, name="Alice", password="secret") + resource = UserResource(model) + attrs = resource.to_attributes() + assert "password" not in attrs + assert attrs["name"] == "Alice" + + def test_non_hidden_fields_present(self): + model = FakeModel(1, name="Alice", email="a@b.com", password="x") + resource = UserResource(model) + attrs = resource.to_attributes() + assert "name" in attrs + assert "email" in attrs + + def test_multiple_hidden_fields(self): + class StrictUser(JsonResource[FakeModel]): + hidden = ["password", "remember_token", "api_key"] + + model = FakeModel(1, name="Bob", password="s", remember_token="t", api_key="k") + resource = StrictUser(model) + attrs = resource.to_attributes() + assert attrs == {"id": 1, "name": "Bob"} + + def test_id_present_in_attributes_by_default(self): + # 'id' is included in attributes — it is only hidden when explicitly added to hidden=[] + model = FakeModel(7, title="Hi") + resource = PostResource(model) + attrs = resource.to_attributes() + assert "id" in attrs + + def test_id_hidden_when_in_hidden_list(self): + class NoIdAttr(JsonResource[FakeModel]): + hidden = ["id"] + + model = FakeModel(7, title="Hi") + resource = NoIdAttr(model) + attrs = resource.to_attributes() + assert "id" not in attrs + + +# --------------------------------------------------------------------------- +# 5. with_() merges into top-level envelope +# --------------------------------------------------------------------------- + + +class TestWith: + def test_with_adds_top_level_key(self): + model = FakeModel(1, title="Hi") + doc = ArticleResource(model).serialize() + assert "meta" in doc + assert doc["meta"] == {"version": "1.0"} + + def test_with_overrides_to_meta(self): + class PriorityResource(JsonResource[FakeModel]): + def to_meta(self): + return {"from_to_meta": True} + + def with_(self): + # with_() is applied after to_meta(), so it wins + return {"meta": {"from_with_": True}} + + model = FakeModel(1, title="Hi") + doc = PriorityResource(model).serialize() + assert doc["meta"] == {"from_with_": True} + + def test_empty_with_does_not_pollute_document(self): + model = FakeModel(1, title="Hi") + doc = PostResource(model).serialize() + # with_() returns {} by default -> no extra keys + assert set(doc.keys()) == {"data"} + + def test_with_merges_multiple_keys(self): + class RichResource(JsonResource[FakeModel]): + def with_(self): + return {"jsonapi": {"version": "1.1"}, "meta": {"page": 1}} + + model = FakeModel(1, title="Hi") + doc = RichResource(model).serialize() + assert doc["jsonapi"] == {"version": "1.1"} + assert doc["meta"] == {"page": 1} + + +# --------------------------------------------------------------------------- +# 6. JsonResource.collection() — plain list +# --------------------------------------------------------------------------- + + +class TestCollectionList: + def test_returns_resource_collection(self): + models = [FakeModel(i, title=f"Post {i}") for i in range(3)] + coll = PostResource.collection(models) + assert isinstance(coll, ResourceCollection) + + def test_data_length(self): + models = [FakeModel(i, title=f"Post {i}") for i in range(5)] + doc = PostResource.collection(models).serialize() + assert len(doc["data"]) == 5 + + def test_each_item_type(self): + models = [FakeModel(1, title="A"), FakeModel(2, title="B")] + doc = PostResource.collection(models).serialize() + for item in doc["data"]: + assert item["type"] == "posts" + + def test_empty_list(self): + doc = PostResource.collection([]).serialize() + assert doc["data"] == [] + + def test_no_meta_for_plain_list(self): + models = [FakeModel(1, title="A")] + doc = PostResource.collection(models).serialize() + assert "meta" not in doc + + +# --------------------------------------------------------------------------- +# 7. JsonResource.collection() — LengthAwarePaginator +# --------------------------------------------------------------------------- + + +class TestCollectionLengthAwarePaginator: + def _make_paginator(self, n=3, per_page=2, page=1, total=10): + rows = [FakeModel(i, title=f"P{i}") for i in range(1, n + 1)] + collection = _FakeModelCollection(rows) + return LengthAwarePaginator(collection, per_page=per_page, current_page=page, total=total) + + def test_returns_resource_collection(self): + pag = self._make_paginator() + coll = PostResource.collection(pag) + assert isinstance(coll, ResourceCollection) + + def test_items_wrapped(self): + pag = self._make_paginator(n=3) + doc = PostResource.collection(pag).serialize() + assert len(doc["data"]) == 3 + + def test_meta_total(self): + pag = self._make_paginator(n=2, per_page=2, page=1, total=10) + doc = PostResource.collection(pag).serialize() + assert "meta" in doc + assert doc["meta"]["total"] == 10 + + def test_meta_current_page(self): + pag = self._make_paginator(n=2, per_page=2, page=2, total=10) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["current_page"] == 2 + + def test_meta_last_page(self): + pag = self._make_paginator(n=2, per_page=2, page=1, total=10) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["last_page"] == 5 # ceil(10/2) + + def test_meta_next_page(self): + pag = self._make_paginator(n=2, per_page=2, page=1, total=10) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["next_page"] == 2 + + def test_meta_no_next_page_on_last(self): + pag = self._make_paginator(n=2, per_page=2, page=5, total=10) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["next_page"] is None + + +# --------------------------------------------------------------------------- +# 8. JsonResource.collection() — SimplePaginator +# --------------------------------------------------------------------------- + + +class TestCollectionSimplePaginator: + def _make_paginator(self, n=3, per_page=2, page=1): + # SimplePaginator needs n+1 to detect has_more + rows = [FakeModel(i, title=f"P{i}") for i in range(1, n + 2)] + collection = _FakeModelCollection(rows) + return SimplePaginator(collection, per_page=per_page, current_page=page) + + def test_returns_resource_collection(self): + pag = self._make_paginator() + coll = PostResource.collection(pag) + assert isinstance(coll, ResourceCollection) + + def test_meta_present(self): + pag = self._make_paginator(n=3, per_page=2, page=1) + doc = PostResource.collection(pag).serialize() + assert "meta" in doc + + def test_meta_has_next_page(self): + pag = self._make_paginator(n=3, per_page=2, page=1) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["next_page"] == 2 + + def test_meta_current_page(self): + pag = self._make_paginator(n=3, per_page=2, page=3) + doc = PostResource.collection(pag).serialize() + assert doc["meta"]["current_page"] == 3 + + +# --------------------------------------------------------------------------- +# 9. Full serialize() document shape +# --------------------------------------------------------------------------- + + +# --------------------------------------------------------------------------- +# 8b. fields() dot-notation +# --------------------------------------------------------------------------- + + +class TestFieldsDotNotation: + def test_plain_field_restricts_primary_type(self): + model = FakeModel(1, title="Hi", body="World") + doc = PostResource(model).fields("title").serialize() + assert doc["data"]["attributes"] == {"title": "Hi"} + + def test_multiple_plain_fields(self): + model = FakeModel(1, title="Hi", body="World") + doc = PostResource(model).fields("title", "body").serialize() + assert set(doc["data"]["attributes"].keys()) == {"title", "body"} + + def test_dotted_field_restricts_related_type(self): + class AuthorResource(JsonResource[FakeModel]): + type = "users" # explicit type so "users.name" filter matches + + class BlogPostResource(JsonResource[FakeModel]): + def to_relationships(self): + return {"author": AuthorResource(FakeModel(5, name="Alice", email="a@b.com"))} + + model = FakeModel(1, title="Hi") + doc = BlogPostResource(model).include("author").fields("users.name").serialize() + inc = doc["included"][0] + assert inc["attributes"] == {"name": "Alice"} + assert "email" not in inc["attributes"] + + def test_mixed_plain_and_dotted_in_single_call(self): + class AuthorResource(JsonResource[FakeModel]): + type = "users" + + class BlogPostResource(JsonResource[FakeModel]): + def to_relationships(self): + return {"author": AuthorResource(FakeModel(5, name="Alice", email="a@b.com"))} + + model = FakeModel(1, title="Hi", body="World") + doc = BlogPostResource(model).include("author").fields("title", "users.name").serialize() + assert doc["data"]["attributes"] == {"title": "Hi"} + inc = doc["included"][0] + assert inc["attributes"] == {"name": "Alice"} + + def test_unrestricted_type_unaffected(self): + # Restricting "users" fields should not affect "posts" attributes + model = FakeModel(1, title="Hi", body="World") + doc = PostResource(model).fields("users.name").serialize() + assert {"title", "body"}.issubset(doc["data"]["attributes"].keys()) + + def test_collection_plain_field_uses_primary_type(self): + models = [FakeModel(1, title="A"), FakeModel(2, title="B")] + doc = PostResource.collection(models).fields("title").serialize() + for item in doc["data"]: + assert list(item["attributes"].keys()) == ["title"] + + def test_fields_accumulates_across_multiple_calls(self): + model = FakeModel(1, title="Hi", body="World") + doc = PostResource(model).fields("title").fields("body").serialize() + assert set(doc["data"]["attributes"].keys()) == {"title", "body"} + + +class TestSerializeDocumentShape: + def test_data_type_and_id(self): + model = FakeModel(42, title="Foo") + doc = PostResource(model).serialize() + assert doc["data"]["type"] == "posts" + assert doc["data"]["id"] == "42" + + def test_no_extra_keys_by_default(self): + model = FakeModel(1, title="Hi") + doc = PostResource(model).serialize() + assert set(doc.keys()) == {"data"} + + def test_links_present_when_overridden(self): + class LinkedPost(JsonResource[FakeModel]): + def to_links(self): + return {"self": f"/api/posts/{self.id}"} + + model = FakeModel(5, title="Hi") + doc = LinkedPost(model).serialize() + assert doc["links"] == {"self": "/api/posts/5"} + + def test_meta_present_when_overridden(self): + class MetaPost(JsonResource[FakeModel]): + def to_meta(self): + return {"foo": "bar"} + + model = FakeModel(1, title="Hi") + doc = MetaPost(model).serialize() + assert doc["meta"] == {"foo": "bar"} diff --git a/fastapi_startkit/tests/jsonapi/test_json_resource_orm.py b/fastapi_startkit/tests/jsonapi/test_json_resource_orm.py new file mode 100644 index 00000000..5ecb7703 --- /dev/null +++ b/fastapi_startkit/tests/jsonapi/test_json_resource_orm.py @@ -0,0 +1,300 @@ +"""Integration tests for JsonResource with real ORM models. + +Uses the SQLite test database (same fixtures as the masoniteorm test suite) +to verify that JsonResource works end-to-end with actual Model instances: +auto-serialization from model.serialize(), hidden fields, collection() with +real query results, and collection() with LengthAwarePaginator / SimplePaginator. +""" + +from unittest import IsolatedAsyncioTestCase + +from fastapi_startkit.jsonapi import JsonResource, ResourceCollection +from fastapi_startkit.masoniteorm import Model +from fastapi_startkit.masoniteorm.testing.transaction import RefreshDatabase + +from tests.masoniteorm.sqlite.fixtures.db import DB +from tests.masoniteorm.fixtures.migration import migrate, wipe +from tests.masoniteorm.fixtures.seeder import seeder +from tests.masoniteorm.fixtures.model import User, Articles + + +# --------------------------------------------------------------------------- +# Resource definitions +# --------------------------------------------------------------------------- + + +class UserResource(JsonResource[User]): + """Exposes all User fields except sensitive ones.""" + + hidden = ["email_verified_at", "preferences", "address"] + + +class ArticleResource(JsonResource[Articles]): + """Exposes Article fields; author relationship via class-reference form. + + The key "author" tells the framework to read ``self.model.author`` + and wrap it automatically with ``UserResource``. + """ + + def to_relationships(self): + return {"author": UserResource} + + +# --------------------------------------------------------------------------- +# Test base — mirrors tests/masoniteorm/sqlite/test_case.py +# --------------------------------------------------------------------------- + + +class OrmTestCase(RefreshDatabase, IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.db = DB + Model.db_manager = DB + self.schema = DB.get_schema_builder() + await self._reset_db() + + async def asyncTearDown(self): + await DB.clear() + await wipe(DB.get_schema_builder()) + + @staticmethod + async def _reset_db(): + await DB.clear() + schema = DB.get_schema_builder() + await wipe(schema) + await migrate(schema) + await seeder() + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestJsonResourceWithOrmModel(OrmTestCase): + """JsonResource auto-serializes directly from a real Model instance.""" + + async def test_type_derived_from_class_name(self): + user = await User.first() + resource = UserResource(user) + self.assertEqual(resource.type, "users") + + async def test_id_set_from_model(self): + user = await User.first() + resource = UserResource(user) + self.assertEqual(resource.id, user.id) + + async def test_attributes_come_from_model_serialize(self): + user = await User.first() + doc = UserResource(user).serialize() + attrs = doc["data"]["attributes"] + + # Fields from User.serialize() must appear in attributes + self.assertIn("name", attrs) + self.assertIn("email", attrs) + self.assertIn("is_admin", attrs) + + async def test_hidden_fields_excluded(self): + user = await User.first() + doc = UserResource(user).serialize() + attrs = doc["data"]["attributes"] + + self.assertNotIn("email_verified_at", attrs) + self.assertNotIn("preferences", attrs) + self.assertNotIn("address", attrs) + + async def test_non_hidden_fields_present(self): + user = await User.first() + doc = UserResource(user).serialize() + attrs = doc["data"]["attributes"] + + self.assertEqual(attrs["name"], "Joe") + self.assertEqual(attrs["email"], "admin@admin.com") + self.assertTrue(attrs["is_admin"]) + + async def test_fields_chain_restricts_attributes(self): + user = await User.first() + doc = UserResource(user).fields("name").serialize() + attrs = doc["data"]["attributes"] + + self.assertIn("name", attrs) + self.assertNotIn("email", attrs) + self.assertNotIn("is_admin", attrs) + + async def test_data_envelope_shape(self): + user = await User.first() + doc = UserResource(user).serialize() + + self.assertIn("data", doc) + self.assertEqual(doc["data"]["type"], "users") + self.assertEqual(doc["data"]["id"], str(user.id)) + self.assertIn("attributes", doc["data"]) + + +class TestJsonResourceCollectionWithOrm(OrmTestCase): + """JsonResource.collection() wraps real ORM query results.""" + + async def test_collection_from_all(self): + users = await User.all() + doc = UserResource.collection(users).serialize() + + self.assertIsInstance(doc["data"], list) + self.assertEqual(len(doc["data"]), 2) # seeder creates 2 users + + async def test_collection_returns_resource_collection(self): + users = await User.all() + coll = UserResource.collection(users) + self.assertIsInstance(coll, ResourceCollection) + + async def test_collection_items_have_correct_type(self): + users = await User.all() + doc = UserResource.collection(users).serialize() + + for item in doc["data"]: + self.assertEqual(item["type"], "users") + + async def test_collection_fields_chain(self): + users = await User.all() + doc = UserResource.collection(users).fields("name", "email").serialize() + + for item in doc["data"]: + self.assertIn("name", item["attributes"]) + self.assertIn("email", item["attributes"]) + self.assertNotIn("is_admin", item["attributes"]) + + async def test_no_meta_for_plain_list(self): + users = await User.all() + doc = UserResource.collection(users).serialize() + self.assertNotIn("meta", doc) + + +class TestJsonResourceWithLengthAwarePaginator(OrmTestCase): + """collection() with a real LengthAwarePaginator includes pagination meta.""" + + async def test_returns_resource_collection(self): + paginator = await User.query().paginate(1) + coll = UserResource.collection(paginator) + self.assertIsInstance(coll, ResourceCollection) + + async def test_data_contains_items(self): + paginator = await User.query().paginate(1) + doc = UserResource.collection(paginator).serialize() + self.assertIsInstance(doc["data"], list) + self.assertGreater(len(doc["data"]), 0) + + async def test_meta_total_present(self): + paginator = await User.query().paginate(1) + doc = UserResource.collection(paginator).serialize() + self.assertIn("meta", doc) + self.assertEqual(doc["meta"]["total"], 2) + + async def test_meta_per_page(self): + paginator = await User.query().paginate(1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["per_page"], 1) + + async def test_meta_current_page(self): + paginator = await User.query().paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["current_page"], 1) + + async def test_meta_last_page(self): + paginator = await User.query().paginate(1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["last_page"], 2) # 2 users, 1 per page + + async def test_meta_next_page(self): + paginator = await User.query().paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["next_page"], 2) + + async def test_meta_no_next_page_on_last(self): + paginator = await User.query().paginate(1, 2) # page 2 of 2 + doc = UserResource.collection(paginator).serialize() + self.assertIsNone(doc["meta"]["next_page"]) + + async def test_fields_chain_on_paginated_collection(self): + paginator = await User.query().paginate(10) + doc = UserResource.collection(paginator).fields("name").serialize() + for item in doc["data"]: + self.assertIn("name", item["attributes"]) + self.assertNotIn("email", item["attributes"]) + + +class TestJsonResourceWithSimplePaginator(OrmTestCase): + """collection() with a real SimplePaginator includes pagination meta.""" + + async def test_returns_resource_collection(self): + paginator = await User.query().simple_paginate(1, 1) + coll = UserResource.collection(paginator) + self.assertIsInstance(coll, ResourceCollection) + + async def test_meta_present(self): + paginator = await User.query().simple_paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertIn("meta", doc) + + async def test_meta_current_page(self): + paginator = await User.query().simple_paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["current_page"], 1) + + async def test_meta_has_next_page(self): + paginator = await User.query().simple_paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertEqual(doc["meta"]["next_page"], 2) + + async def test_meta_no_total(self): + # SimplePaginator does not provide total / last_page + paginator = await User.query().simple_paginate(1, 1) + doc = UserResource.collection(paginator).serialize() + self.assertNotIn("total", doc["meta"]) + self.assertNotIn("last_page", doc["meta"]) + + +class TestJsonResourceRelationshipsWithOrm(OrmTestCase): + """to_relationships() wired to real ORM data via with_().""" + + async def test_relationships_in_data_when_author_set(self): + article = await Articles.first() + author = await User.find(article.user_id) + + # Attach author onto model so to_relationships() can access it + article.author = author + doc = ArticleResource(article).serialize() + + self.assertIn("relationships", doc["data"]) + self.assertIn("author", doc["data"]["relationships"]) + + async def test_include_author_sideloaded(self): + article = await Articles.first() + author = await User.find(article.user_id) + article.author = author + + doc = ArticleResource(article).include("author").serialize() + + self.assertIn("included", doc) + inc = doc["included"][0] + self.assertEqual(inc["type"], "users") + self.assertEqual(inc["id"], str(author.id)) + + async def test_include_author_attributes(self): + article = await Articles.first() + author = await User.find(article.user_id) + article.author = author + + doc = ArticleResource(article).include("author").serialize() + + inc_attrs = doc["included"][0]["attributes"] + self.assertIn("name", inc_attrs) + self.assertEqual(inc_attrs["name"], "Joe") + + async def test_fields_restrict_included_resource(self): + article = await Articles.first() + author = await User.find(article.user_id) + article.author = author + + doc = ArticleResource(article).include("author").fields("users.name").serialize() + + inc_attrs = doc["included"][0]["attributes"] + self.assertIn("name", inc_attrs) + self.assertNotIn("email", inc_attrs) diff --git a/fastapi_startkit/tests/jsonapi/test_list_response.py b/fastapi_startkit/tests/jsonapi/test_list_response.py index 2b6e6ea4..d2de4f23 100644 --- a/fastapi_startkit/tests/jsonapi/test_list_response.py +++ b/fastapi_startkit/tests/jsonapi/test_list_response.py @@ -1,44 +1,49 @@ -"""Tests for JsonAPIListResponse serialization.""" +"""Tests for ResourceCollection serialization.""" -from fastapi_startkit.jsonapi import JsonAPIListResponse, JsonAPIResponse +from fastapi_startkit.jsonapi import JsonResource, ResourceCollection # --------------------------------------------------------------------------- -# Fixture resources +# Fake models # --------------------------------------------------------------------------- -class AuthorResource(JsonAPIResponse): - type = "authors" - attributes = ["name"] - +class FakeAuthor: def __init__(self, id_, name): self.id = id_ self.name = name - def to_attributes(self): - return {"name": self.name} - + def serialize(self): + return {"id": self.id, "name": self.name} -class ArticleResource(JsonAPIResponse): - type = "articles" - attributes = ["title"] +class FakeArticle: def __init__(self, id_, title, author=None): self.id = id_ self.title = title - self._author = author + self.author = author + + def serialize(self): + return {"id": self.id, "title": self.title} + + +# --------------------------------------------------------------------------- +# Fixture resources +# --------------------------------------------------------------------------- + + +class AuthorResource(JsonResource[FakeAuthor]): + pass # type="authors", auto-attrs - def to_attributes(self): - return {"title": self.title} +class ArticleResource(JsonResource[FakeArticle]): def to_relationships(self): - if self._author is None: + if self.model.author is None: return None - return {"author": self._author} + return {"author": AuthorResource(self.model.author)} -class ArticleListWithMeta(JsonAPIListResponse): +class ArticleListWithMeta(ResourceCollection): def to_meta(self): return {"total": len(self._items)} @@ -53,63 +58,63 @@ def to_links(self): class TestJsonAPIListResponseStructure: def test_data_is_list(self): - items = [ArticleResource(1, "A"), ArticleResource(2, "B")] - doc = JsonAPIListResponse(items).serialize() + items = [ArticleResource(FakeArticle(1, "A")), ArticleResource(FakeArticle(2, "B"))] + doc = ResourceCollection(items).serialize() assert isinstance(doc["data"], list) def test_data_length(self): - items = [ArticleResource(i, f"Article {i}") for i in range(3)] - doc = JsonAPIListResponse(items).serialize() + items = [ArticleResource(FakeArticle(i, f"Article {i}")) for i in range(3)] + doc = ResourceCollection(items).serialize() assert len(doc["data"]) == 3 def test_each_item_has_type_and_id(self): - items = [ArticleResource(1, "A"), ArticleResource(2, "B")] - doc = JsonAPIListResponse(items).serialize() + items = [ArticleResource(FakeArticle(1, "A")), ArticleResource(FakeArticle(2, "B"))] + doc = ResourceCollection(items).serialize() for item_data in doc["data"]: assert "type" in item_data assert "id" in item_data def test_data_types(self): - items = [ArticleResource(1, "A")] - doc = JsonAPIListResponse(items).serialize() + items = [ArticleResource(FakeArticle(1, "A"))] + doc = ResourceCollection(items).serialize() assert doc["data"][0]["type"] == "articles" def test_empty_list(self): - doc = JsonAPIListResponse([]).serialize() + doc = ResourceCollection([]).serialize() assert doc["data"] == [] assert "included" not in doc def test_no_links_by_default(self): - doc = JsonAPIListResponse([ArticleResource(1, "A")]).serialize() + doc = ResourceCollection([ArticleResource(FakeArticle(1, "A"))]).serialize() assert "links" not in doc def test_no_meta_by_default(self): - doc = JsonAPIListResponse([ArticleResource(1, "A")]).serialize() + doc = ResourceCollection([ArticleResource(FakeArticle(1, "A"))]).serialize() assert "meta" not in doc def test_links_when_overridden(self): - items = [ArticleResource(1, "A")] + items = [ArticleResource(FakeArticle(1, "A"))] doc = ArticleListWithMeta(items).serialize() assert doc["links"] == {"self": "/api/articles"} def test_meta_when_overridden(self): - items = [ArticleResource(1, "A"), ArticleResource(2, "B")] + items = [ArticleResource(FakeArticle(1, "A")), ArticleResource(FakeArticle(2, "B"))] doc = ArticleListWithMeta(items).serialize() assert doc["meta"] == {"total": 2} class TestJsonAPIListResponseInclude: def test_included_when_include_specified(self): - author = AuthorResource(5, "Bob") - items = [ArticleResource(1, "A", author=author)] - doc = JsonAPIListResponse(items).serialize(include=["author"]) + author = FakeAuthor(5, "Bob") + items = [ArticleResource(FakeArticle(1, "A", author=author))] + doc = ResourceCollection(items).include("author").serialize() assert "included" in doc assert len(doc["included"]) == 1 def test_included_contains_correct_resource(self): - author = AuthorResource(5, "Bob") - items = [ArticleResource(1, "A", author=author)] - doc = JsonAPIListResponse(items).serialize(include=["author"]) + author = FakeAuthor(5, "Bob") + items = [ArticleResource(FakeArticle(1, "A", author=author))] + doc = ResourceCollection(items).include("author").serialize() inc = doc["included"][0] assert inc["type"] == "authors" assert inc["id"] == "5" @@ -117,34 +122,34 @@ def test_included_contains_correct_resource(self): def test_deduplication_across_items(self): """Same author linked from two articles must appear only once.""" - author = AuthorResource(5, "Bob") + author = FakeAuthor(5, "Bob") items = [ - ArticleResource(1, "A", author=author), - ArticleResource(2, "B", author=author), + ArticleResource(FakeArticle(1, "A", author=author)), + ArticleResource(FakeArticle(2, "B", author=author)), ] - doc = JsonAPIListResponse(items).serialize(include=["author"]) + doc = ResourceCollection(items).include("author").serialize() assert len(doc["included"]) == 1 def test_multiple_distinct_authors_included(self): - alice = AuthorResource(1, "Alice") - bob = AuthorResource(2, "Bob") + alice = FakeAuthor(1, "Alice") + bob = FakeAuthor(2, "Bob") items = [ - ArticleResource(1, "A", author=alice), - ArticleResource(2, "B", author=bob), + ArticleResource(FakeArticle(1, "A", author=alice)), + ArticleResource(FakeArticle(2, "B", author=bob)), ] - doc = JsonAPIListResponse(items).serialize(include=["author"]) + doc = ResourceCollection(items).include("author").serialize() assert len(doc["included"]) == 2 ids = {inc["id"] for inc in doc["included"]} assert ids == {"1", "2"} def test_no_included_when_not_in_include(self): - author = AuthorResource(5, "Bob") - items = [ArticleResource(1, "A", author=author)] - doc = JsonAPIListResponse(items).serialize(include=["comments"]) + author = FakeAuthor(5, "Bob") + items = [ArticleResource(FakeArticle(1, "A", author=author))] + doc = ResourceCollection(items).include("comments").serialize() assert "included" not in doc def test_no_included_on_empty_include(self): - author = AuthorResource(5, "Bob") - items = [ArticleResource(1, "A", author=author)] - doc = JsonAPIListResponse(items).serialize(include=[]) + author = FakeAuthor(5, "Bob") + items = [ArticleResource(FakeArticle(1, "A", author=author))] + doc = ResourceCollection(items).serialize() assert "included" not in doc diff --git a/fastapi_startkit/tests/jsonapi/test_query_params.py b/fastapi_startkit/tests/jsonapi/test_query_params.py index 55246b83..494ef679 100644 --- a/fastapi_startkit/tests/jsonapi/test_query_params.py +++ b/fastapi_startkit/tests/jsonapi/test_query_params.py @@ -1,53 +1,65 @@ """Tests for parse_include(), parse_fields(), sparse fieldsets, and the FastAPI ASGI __call__ integration (return resource directly). + +Sparse fieldsets and sideloading are applied via the fluent chain API: + resource.include("author").fields("posts", ["title"]).serialize() """ import pytest from fastapi_startkit.jsonapi import ( - JsonAPIListResponse, - JsonAPIResponse, + JsonResource, parse_fields, parse_include, ) # --------------------------------------------------------------------------- -# Fixture resources +# Fake models # --------------------------------------------------------------------------- -class UserResource(JsonAPIResponse): - type = "users" - attributes = ["name", "email", "role"] - +class FakeUser: def __init__(self, id_, name, email="u@example.com", role="member"): self.id = id_ self.name = name self.email = email self.role = role - def to_attributes(self): - return {"name": self.name, "email": self.email, "role": self.role} - + def serialize(self): + return {"id": self.id, "name": self.name, "email": self.email, "role": self.role} -class PostResource(JsonAPIResponse): - type = "posts" - attributes = ["title", "body", "created_at"] +class FakePost: def __init__(self, id_, title, body="", created_at="2024-01-01", author=None): self.id = id_ self.title = title self.body = body self.created_at = created_at - self._author = author + self.author = author + + def serialize(self): + return { + "id": self.id, + "title": self.title, + "body": self.body, + "created_at": self.created_at, + } + + +# --------------------------------------------------------------------------- +# Fixture resources +# --------------------------------------------------------------------------- + - def to_attributes(self): - return {"title": self.title, "body": self.body, "created_at": self.created_at} +class UserResource(JsonResource[FakeUser]): + pass # type="users", auto-attrs + +class PostResource(JsonResource[FakePost]): def to_relationships(self): - if self._author is None: + if self.model.author is None: return None - return {"author": self._author} + return {"author": UserResource(self.model.author)} # --------------------------------------------------------------------------- @@ -123,49 +135,41 @@ def test_no_fields_keys_returns_empty(self): class TestSparseFieldsets: def test_single_resource_fields_filtered(self): - post = PostResource(1, "Hello", "World", "2024-01-01") - doc = post.serialize(fields={"posts": ["title"]}) + post = PostResource(FakePost(1, "Hello", "World", "2024-01-01")) + doc = post.fields("title").serialize() assert doc["data"]["attributes"] == {"title": "Hello"} def test_multiple_fields_filtered(self): - post = PostResource(1, "Hello", "World", "2024-01-01") - doc = post.serialize(fields={"posts": ["title", "body"]}) + post = PostResource(FakePost(1, "Hello", "World", "2024-01-01")) + doc = post.fields("title", "body").serialize() assert doc["data"]["attributes"] == {"title": "Hello", "body": "World"} def test_fields_for_other_type_does_not_affect_this_resource(self): - post = PostResource(1, "Hello", "World", "2024-01-01") - doc = post.serialize(fields={"users": ["name"]}) # no "posts" key - # all attributes should be present - assert set(doc["data"]["attributes"].keys()) == {"title", "body", "created_at"} + post = PostResource(FakePost(1, "Hello", "World", "2024-01-01")) + # "users.name" restricts the users type only — posts attributes are unaffected + doc = post.fields("users.name").serialize() + assert {"title", "body", "created_at"}.issubset(doc["data"]["attributes"].keys()) def test_included_resources_also_filtered(self): - author = UserResource(5, "Alice", "alice@example.com", "admin") - post = PostResource(1, "Hello", "World", author=author) - doc = post.serialize( - include=["author"], - fields={"users": ["name"]}, - ) + author = FakeUser(5, "Alice", "alice@example.com", "admin") + post = PostResource(FakePost(1, "Hello", "World", author=author)) + doc = post.include("author").fields("users.name").serialize() inc = next(i for i in doc["included"] if i["type"] == "users") assert inc["attributes"] == {"name": "Alice"} assert "email" not in inc["attributes"] def test_list_response_fields_filtered(self): - posts = [ - PostResource(1, "A", "aa", "2024-01-01"), - PostResource(2, "B", "bb", "2024-01-02"), - ] - doc = JsonAPIListResponse(posts).serialize(fields={"posts": ["title"]}) + models = [FakePost(1, "A", "aa", "2024-01-01"), FakePost(2, "B", "bb", "2024-01-02")] + doc = PostResource.collection(models).fields("title").serialize() for item in doc["data"]: assert list(item["attributes"].keys()) == ["title"] def test_combined_include_and_fields(self): - """include= and fields[*]= should work together.""" - author = UserResource(10, "Bob", "bob@example.com", "editor") - post = PostResource(1, "Post", "Body", author=author) - doc = post.serialize( - include=["author"], - fields={"posts": ["title"], "users": ["name", "email"]}, - ) + """include and fields should work together via chain API.""" + author = FakeUser(10, "Bob", "bob@example.com", "editor") + post = PostResource(FakePost(1, "Post", "Body", author=author)) + # "title" -> posts type (primary); "users.name" and "users.email" -> users type + doc = post.include("author").fields("title", "users.name", "users.email").serialize() assert doc["data"]["attributes"] == {"title": "Post"} inc = doc["included"][0] assert inc["attributes"] == {"name": "Bob", "email": "bob@example.com"} @@ -177,7 +181,7 @@ def test_combined_include_and_fields(self): class TestFastAPIAsgiResponse: - """Test that JsonAPIResponse instances can be returned directly from + """Test that JsonResource instances can be returned directly from FastAPI endpoints (the ASGI __call__ path).""" @pytest.fixture @@ -189,23 +193,25 @@ def app(self): @api.get("/api/users/{id}") async def get_user(id: int): - user = UserResource(id, "Alice", "alice@example.com", "admin") - return user # ← returned directly, no .serialize() call + user = UserResource(FakeUser(id, "Alice", "alice@example.com", "admin")) + return user # <- returned directly, no .serialize() call @api.get("/api/posts/{id}") async def get_post(id: int): - author = UserResource(1, "Alice") - post = PostResource(id, "My Post", "Body text", author=author) + author = FakeUser(1, "Alice") + post = PostResource(FakePost(id, "My Post", "Body text", author=author)) return post @api.get("/api/posts") async def list_posts(): - author = UserResource(1, "Alice") - posts = [ - PostResource(1, "Post One", author=author), - PostResource(2, "Post Two"), - ] - return JsonAPIListResponse(posts) + author = FakeUser(1, "Alice") + posts = PostResource.collection( + [ + FakePost(1, "Post One", author=author), + FakePost(2, "Post Two"), + ] + ) + return posts return TestClient(api) diff --git a/fastapi_startkit/tests/jsonapi/test_response.py b/fastapi_startkit/tests/jsonapi/test_response.py index 6a26340f..b3ef5970 100644 --- a/fastapi_startkit/tests/jsonapi/test_response.py +++ b/fastapi_startkit/tests/jsonapi/test_response.py @@ -1,43 +1,47 @@ -"""Tests for JsonAPIResponse serialization.""" +"""Tests for JsonResource serialization.""" -from fastapi_startkit.jsonapi import JsonAPIResponse +from fastapi_startkit.jsonapi import JsonResource, ResourceCollection # --------------------------------------------------------------------------- -# Fixture resources +# Fake models # --------------------------------------------------------------------------- -class UserResource(JsonAPIResponse): - type = "users" - attributes = ["name", "email"] - +class FakeUser: def __init__(self, id_, name, email="user@example.com"): self.id = id_ self.name = name self.email = email - def to_attributes(self): - return {"name": self.name, "email": self.email} - + def serialize(self): + return {"id": self.id, "name": self.name, "email": self.email} -class PostResource(JsonAPIResponse): - type = "posts" - attributes = ["title", "body"] +class FakePost: def __init__(self, id_, title, body, author=None): self.id = id_ self.title = title self.body = body - self._author = author + self.author = author + + def serialize(self): + return {"id": self.id, "title": self.title, "body": self.body} + + +# --------------------------------------------------------------------------- +# Fixture resources +# --------------------------------------------------------------------------- + + +class UserResource(JsonResource[FakeUser]): + pass # type="users", auto-attrs - def to_attributes(self): - return {"title": self.title, "body": self.body} +class PostResource(JsonResource[FakePost]): + # Class reference — framework auto-wraps self.model.author with UserResource def to_relationships(self): - if self._author is None: - return None - return {"author": self._author} + return {"author": UserResource} class PostWithLinksResource(PostResource): @@ -55,73 +59,75 @@ def to_meta(self): class TestJsonAPIResponseStructure: def test_serialize_returns_dict(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert isinstance(doc, dict) def test_data_key_present(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert "data" in doc def test_data_type(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert doc["data"]["type"] == "posts" def test_data_id_as_string(self): - post = PostResource(42, "Hello", "World") + post = PostResource(FakePost(42, "Hello", "World")) doc = post.serialize() assert doc["data"]["id"] == "42" def test_data_attributes(self): - post = PostResource(1, "My Title", "My Body") + post = PostResource(FakePost(1, "My Title", "My Body")) doc = post.serialize() - assert doc["data"]["attributes"] == {"title": "My Title", "body": "My Body"} + attrs = doc["data"]["attributes"] + assert attrs["title"] == "My Title" + assert attrs["body"] == "My Body" def test_no_relationships_key_when_none(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert "relationships" not in doc["data"] def test_no_included_key_when_no_include(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) doc = post.serialize() assert "included" not in doc def test_no_links_key_when_none(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert "links" not in doc def test_no_meta_key_when_none(self): - post = PostResource(1, "Hello", "World") + post = PostResource(FakePost(1, "Hello", "World")) doc = post.serialize() assert "meta" not in doc def test_links_present_when_overridden(self): - post = PostWithLinksResource(5, "Hello", "World") + post = PostWithLinksResource(FakePost(5, "Hello", "World")) doc = post.serialize() assert doc["links"] == {"self": "/api/posts/5"} def test_meta_present_when_overridden(self): - post = PostWithLinksResource(5, "Hello", "World") + post = PostWithLinksResource(FakePost(5, "Hello", "World")) doc = post.serialize() assert doc["meta"] == {"version": 1} class TestJsonAPIResponseRelationships: def test_relationships_in_data_when_present(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) doc = post.serialize() assert "relationships" in doc["data"] assert "author" in doc["data"]["relationships"] def test_relationship_data_shape(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) doc = post.serialize() rel = doc["data"]["relationships"]["author"]["data"] assert rel == {"type": "users", "id": "10"} @@ -129,15 +135,15 @@ def test_relationship_data_shape(self): class TestJsonAPIResponseInclude: def test_included_when_include_specified(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) - doc = post.serialize(include=["author"]) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) + doc = post.include("author").serialize() assert "included" in doc def test_included_contains_author(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) - doc = post.serialize(include=["author"]) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) + doc = post.include("author").serialize() included = doc["included"] assert len(included) == 1 assert included[0]["type"] == "users" @@ -145,52 +151,150 @@ def test_included_contains_author(self): assert included[0]["attributes"]["name"] == "Alice" def test_no_included_when_rel_not_in_include(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) - doc = post.serialize(include=["comments"]) # 'comments' does not exist + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) + doc = post.include("comments").serialize() # 'comments' does not exist assert "included" not in doc def test_no_included_key_when_include_is_empty_list(self): - author = UserResource(10, "Alice") - post = PostResource(1, "Hello", "World", author=author) - doc = post.serialize(include=[]) + author = FakeUser(10, "Alice") + post = PostResource(FakePost(1, "Hello", "World", author=author)) + doc = post.serialize() assert "included" not in doc def test_no_duplicate_included_across_two_posts_same_author(self): """Duplicates should not appear even if processed by list response.""" - from fastapi_startkit.jsonapi import JsonAPIListResponse - - author = UserResource(10, "Alice") - post1 = PostResource(1, "P1", "B1", author=author) - post2 = PostResource(2, "P2", "B2", author=author) - doc = JsonAPIListResponse([post1, post2]).serialize(include=["author"]) + author = FakeUser(10, "Alice") + post1 = PostResource(FakePost(1, "P1", "B1", author=author)) + post2 = PostResource(FakePost(2, "P2", "B2", author=author)) + doc = ResourceCollection([post1, post2]).include("author").serialize() assert len(doc["included"]) == 1 -class TestJsonAPIResponseDefaultAttributes: - """Test the default to_attributes() logic when attributes is a list.""" +class TestRelationshipForms: + """to_relationships() accepts three equivalent value forms.""" + + def test_class_reference_auto_wraps_model_attribute(self): + """``UserResource`` → framework reads model.author and wraps it.""" - def test_default_to_attributes_reads_instance_attrs(self): - class TagResource(JsonAPIResponse): - type = "tags" - attributes = ["name", "slug"] + class AuthorResource(JsonResource[FakeUser]): + pass + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + return {"author": AuthorResource} + + author = FakeUser(5, "Bob") + post = FakePost(1, "Hi", "There", author=author) + doc = ArticleResource(post).serialize() + assert "author" in doc["data"]["relationships"] + assert doc["data"]["relationships"]["author"]["data"]["id"] == "5" + + def test_class_reference_none_when_attribute_missing(self): + """Class reference skips the key when model attribute is None/absent.""" + + class AuthorResource(JsonResource[FakeUser]): + pass + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + return {"author": AuthorResource} + + post = FakePost(1, "Hi", "There", author=None) + doc = ArticleResource(post).serialize() + assert "relationships" not in doc["data"] - def __init__(self, id_, name, slug): + def test_lambda_callable(self): + """Lambda is called with no args; its return value is used directly.""" + + class AuthorResource(JsonResource[FakeUser]): + pass + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + return { + "author": lambda: AuthorResource(self.model.author), + } + + author = FakeUser(7, "Carol") + post = FakePost(1, "Hi", "There", author=author) + doc = ArticleResource(post).serialize() + assert doc["data"]["relationships"]["author"]["data"]["id"] == "7" + + def test_explicit_instance(self): + """Passing an already-constructed JsonResource instance works too.""" + + class AuthorResource(JsonResource[FakeUser]): + pass + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + return {"author": AuthorResource(self.model.author)} + + author = FakeUser(9, "Dave") + post = FakePost(1, "Hi", "There", author=author) + doc = ArticleResource(post).serialize() + assert doc["data"]["relationships"]["author"]["data"]["id"] == "9" + + def test_collection_relationship_produces_array_linkage(self): + """A ResourceCollection value produces ``{"data": [...]}`` linkage.""" + + class CommentResource(JsonResource[FakeUser]): + type = "comments" + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + comments = [ + CommentResource(FakeUser(1, "c1")), + CommentResource(FakeUser(2, "c2")), + ] + return {"comments": ResourceCollection(comments)} + + post = FakePost(1, "Hi", "There") + doc = ArticleResource(post).serialize() + linkage = doc["data"]["relationships"]["comments"]["data"] + assert isinstance(linkage, list) + assert len(linkage) == 2 + assert linkage[0] == {"type": "comments", "id": "1"} + + def test_collection_relationship_included_when_sideloaded(self): + """include("comments") sideloads all items from a collection relationship.""" + + class CommentResource(JsonResource[FakeUser]): + type = "comments" + + class ArticleResource(JsonResource[FakePost]): + def to_relationships(self): + comments = [ + CommentResource(FakeUser(1, "c1")), + CommentResource(FakeUser(2, "c2")), + ] + return {"comments": ResourceCollection(comments)} + + post = FakePost(1, "Hi", "There") + doc = ArticleResource(post).include("comments").serialize() + assert len(doc["included"]) == 2 + assert all(i["type"] == "comments" for i in doc["included"]) + + def test_lambda_using_collection(self): + """Lambda calls ResourceClass.collection() for has-many relationships.""" + + class FakePostWithComments: + def __init__(self, id_, comments): self.id = id_ - self.name = name - self.slug = slug + self.comments = comments - tag = TagResource(1, "Python", "python") - doc = tag.serialize() - assert doc["data"]["attributes"] == {"name": "Python", "slug": "python"} + def serialize(self): + return {"id": self.id} - def test_default_to_attributes_returns_none_when_no_attrs(self): - class EmptyResource(JsonAPIResponse): - type = "empty" + class CommentResource(JsonResource[FakeUser]): + type = "comments" - def __init__(self): - self.id = 1 + class ArticleResource(JsonResource[FakePostWithComments]): + def to_relationships(self): + return {"comments": lambda: CommentResource.collection(self.model.comments)} - resource = EmptyResource() - doc = resource.serialize() - assert "attributes" not in doc["data"] + comments = [FakeUser(1, "c1"), FakeUser(2, "c2")] + post = FakePostWithComments(1, comments) + doc = ArticleResource(post).include("comments").serialize() + assert len(doc["included"]) == 2