API Reference

class ActiveDocument(BaseModel):
    """Experimental ORM class for Strapi document."""

    __plural_api_id__: ClassVar[str]
    __managed_fields__: ClassVar[set[str]] = {
        "id",
        "document_id",
        "created_at",
        "updated_at",
        "published_at",
        "__plural_api_id__",
        "__managed_fields__",
    }
    _relations_populated: bool = PrivateAttr(default=False)
    id: int | None = DocumentField(default=None, unique=True)
    document_id: str | None = DocumentField(default=None, alias="documentId", unique=True)
    created_at: datetime.datetime | None = Field(default=None, alias="createdAt")
    updated_at: datetime.datetime | None = Field(default=None, alias="updatedAt")
    published_at: datetime.datetime | None = Field(default=None, alias="publishedAt")

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if "__plural_api_id__" not in cls.__dict__:
            cls.__plural_api_id__ = cls.__name__.lower() + "s"

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
        document_id: str,
        populate_all: bool = True,
    ) -> Self:
        """Get document by document id."""
        response = await client.get_document(
            plural_api_id=cls.__plural_api_id__,
            document_id=document_id,
            populate=list(cls._get_relation_fields()) if populate_all else None,
        )
        return cls.model_validate(response.data)

    @classmethod
    async def get_documents(
        cls,
        client: StrapiClientAsync,
        populate_all: bool = True,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = 0,
        page: int | None = None,
        limit: int = 25,
        with_count: bool = True,
    ) -> list[Self]:
        """Get list of documents."""
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=list(cls._get_relation_fields()) if populate_all else None,
            fields=list(cls._get_document_fields(with_relations=False)),
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        documents = [cls.model_validate(document) for document in response.data]
        if populate_all:
            for document in documents:
                document.set_relations_populated(populate_all)
        return documents

    async def create_document(self, client: StrapiClientAsync) -> Self:
        """Create new document from object."""
        response = await client.create_document(plural_api_id=self.__plural_api_id__, data=self.model_dump_variable())
        return self.model_validate(response.data)

    async def update_document(self, client: StrapiClientAsync) -> Self:
        """Update document fields from object."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to update document")
        if not self.relations_populated():
            warnings.warn(
                "Some relations are not populated, so all relations will not be updated. Use refresh() method to populate relations."
            )
            # TODO: relations: set, connect, disconnect. Preserve hashing. Check warning
            data = self.model_dump_variable(exclude=self._get_relation_fields())
        else:
            data = self.model_dump_variable()
        response = await client.update_document(
            plural_api_id=self.__plural_api_id__, document_id=self.document_id, data=data
        )
        return self.model_validate(response.data)

    async def delete_document(self, client: StrapiClientAsync) -> None:
        """Delete document attached to object."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to delete document")
        await client.delete_document(plural_api_id=self.__plural_api_id__, document_id=self.document_id)

    def relations_populated(self) -> bool:
        return self._relations_populated or not self._get_relation_fields()

    def set_relations_populated(self, value: bool) -> Self:
        self._relations_populated = value
        return self

    async def refresh(self, client: StrapiClientAsync, populate_all: bool = True) -> Self:
        """Refresh object with latest data from Strapi including relations."""
        if not self.document_id:
            raise RuntimeError("Document ID cannot be empty to refresh object")
        document = await self.__class__.get_document(client, self.document_id, populate_all=populate_all)
        for field in document.model_fields:
            setattr(self, field, getattr(document, field))

        self.set_relations_populated(populate_all)
        return self

    async def upsert_document(self, client: StrapiClientAsync) -> Self:
        """Create document or update fields."""
        keys: set[str] = self._unique_fields - self.__managed_fields__
        if not keys:
            raise RuntimeError("For upsert at least one model field should be declared as unique")
        model_dict = self.model_dump_variable()
        filters = {key: {"$eq": model_dict[key]} for key in keys}
        cur_response = await client.get_documents(
            plural_api_id=self.__plural_api_id__,
            filters=filters,
            fields=list(self._get_document_fields()),
            start=0,
            batch_size=1,
            with_count=True,
        )
        total_count: int = cur_response.meta.get_total_count()
        if total_count > 1:
            raise RuntimeError(f"Keys are ambiguous, found {total_count} records")
        elif total_count == 0:
            return await self.create_document(client)
        else:
            cur_document = self.model_validate(cur_response.data[0])
            if cur_document.model_hash() == self.model_hash():
                return cur_document
            else:
                self.id = cur_document.id
                self.document_id = cur_document.document_id
                return await self.update_document(client)

    def model_dump_variable(self, exclude: set[str] | None = None) -> dict[str, Any]:
        exclude = exclude or set()
        model_dict = self.model_dump(by_alias=True, exclude=self.__managed_fields__ | exclude)
        for rel in self._get_relation_fields():
            if model_dict.get(rel):
                model_dict[rel] = {"set": model_dict[rel]["documentId"]}
        return model_dict

    def model_hash(self) -> str:
        dumped_str = json.dumps(self.model_dump_variable(), sort_keys=True, default=str)
        return hashlib.sha256(dumped_str.encode("utf-8")).hexdigest()

    @property
    def _unique_fields(self) -> set[str]:
        return {f for f, info in self.model_fields.items() if any(m.get("unique", False) for m in info.metadata)}

    @classmethod
    def _get_document_fields(cls, with_relations: bool = True) -> set[str]:
        return {
            info.alias or f
            for f, info in cls.__pydantic_fields__.items()
            if with_relations or not any(m.get("relation", False) for m in info.metadata)
        }

    @classmethod
    def _get_relation_fields(cls) -> set[str]:
        return {
            info.alias or f
            for f, info in cls.__pydantic_fields__.items()
            if any(m.get("relation", False) for m in info.metadata)
        }

class BaseComponent(BasePopulatable):
    """Strapi component."""

class BaseDocument(BasePopulatable):
    """Strapi document with standard fields."""

    id: int
    document_id: str = Field(alias="documentId")
    created_at: datetime.datetime = Field(alias="createdAt")
    updated_at: datetime.datetime = Field(alias="updatedAt")
    published_at: datetime.datetime = Field(alias="publishedAt")

    @classmethod
    def from_scalar_response(cls, response: DocumentResponse) -> Self:
        return cls.model_validate(response.data)

    @classmethod
    def from_list_response(cls, response: DocumentsResponse) -> list[Self]:
        return [cls.model_validate(d) for d in response.data]

    @classmethod
    def first_from_list_response(cls, response: DocumentsResponse) -> Self | None:
        return cls.model_validate(response.data[0]) if len(response.data) > 0 else None

class LocalizedSmartDocument(SmartDocument):
    """SmartDocument for Strapi content types with internationalization enabled."""

    __abstract_smart_document__: ClassVar[bool] = True
    locale: str

    __managed_fields__: ClassVar[set[str]] = SmartDocument.__managed_fields__ | {"locale"}

class MediaImageDocument(BaseDocument):
    name: str
    alternative_text: str | None = Field(default=None, alias="alternativeText")
    caption: str | None = None
    width: int | None
    height: int | None
    formats: MediaImageFormats | None
    hash: str
    ext: str
    mime: str
    size: float = Field(description="Size in kilobytes")
    url: str
    preview_url: str | None = Field(default=None, alias="previewUrl")
    provider: str
    provider_metadata: dict[str, Any] | None = None

    @property
    def largest_format(self) -> MediaImageFormatVariant | None:
        return self.formats.largest if self.formats else None

class SingleSmartDocument(BaseDocument):
    """Fully automated ORM class for single Strapi document."""

    __single_api_id__: ClassVar[str]

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if "__single_api_id__" not in cls.__dict__:
            name = re.sub(r"(?<!^)(?=[A-Z])", "-", cls.__name__).lower()
            cls.__single_api_id__ = name

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
    ) -> Self:
        """Get single document by single api id."""
        response = await client.get_single_document(single_api_id=cls.__single_api_id__)
        return cls.from_scalar_response(response)

class SmartDocument(BaseDocument):
    """Fully automated ORM class for Strapi documents."""

    __singular_api_id__: ClassVar[str]
    __plural_api_id__: ClassVar[str]
    __content_type_id__: ClassVar[str]
    __abstract_smart_document__: ClassVar[bool]
    __managed_fields__: ClassVar[set[str]] = {
        "id",
        "document_id",
        "created_at",
        "updated_at",
        "published_at",
        "__singular_api_id__",
        "__plural_api_id__",
        "__content_type_id__",
        "__managed_fields__",
    }

    @classmethod
    def __pydantic_init_subclass__(cls, **kwargs: Any) -> None:
        super().__pydantic_init_subclass__(**kwargs)
        if cls.__dict__.get("__abstract_smart_document__", False):
            return
        if "__singular_api_id__" not in cls.__dict__:
            if hasattr(cls, "__singular_api_id__"):
                cls.__singular_api_id__ = cls.__singular_api_id__
            else:
                cls.__singular_api_id__ = re.sub(r"(?<!^)(?=[A-Z])", "-", cls.__name__).lower()
        if "__plural_api_id__" not in cls.__dict__:
            if hasattr(cls, "__plural_api_id__"):
                cls.__plural_api_id__ = cls.__plural_api_id__
            else:
                cls.__plural_api_id__ = f"{cls.__singular_api_id__}s"
        if "__content_type_id__" not in cls.__dict__:
            if hasattr(cls, "__content_type_id__"):
                cls.__content_type_id__ = cls.__content_type_id__
            else:
                cls.__content_type_id__ = f"api::{cls.__singular_api_id__}.{cls.__singular_api_id__}"

    @classmethod
    async def get_document(
        cls,
        client: StrapiClientAsync,
        document_id: str,
        locale: str | None = None,
    ) -> Self:
        """Get document by document id."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_document(
            plural_api_id=cls.__plural_api_id__,
            document_id=document_id,
            fields=fields,
            populate=populate,
            locale=locale,
        )
        return cls.from_scalar_response(response)

    @classmethod
    async def get_documents(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        limit: int = 100,
        with_count: bool = True,
    ) -> list[Self]:
        """Get list of documents."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort or ["id"],
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        return cls.from_list_response(response)

    @classmethod
    async def get_documents_with_meta(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
        start: int | None = None,
        page: int | None = None,
        limit: int = 100,
        with_count: bool = True,
    ) -> tuple[list[Self], ResponseMeta]:
        """Get list of documents."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            page=page,
            start=start,
            batch_size=limit,
            with_count=with_count,
        )
        return cls.from_list_response(response), response.meta

    @classmethod
    async def get_first_document(
        cls,
        client: StrapiClientAsync,
        sort: list[str] | None = None,
        filters: dict[str, Any] | None = None,
        publication_state: str | None = None,
        locale: str | None = None,
    ) -> Self | None:
        """First documents if available."""
        fields, populate = get_model_fields_and_population(cls)
        response = await client.get_documents(
            plural_api_id=cls.__plural_api_id__,
            sort=sort,
            filters=filters,
            populate=populate,
            fields=fields,
            publication_state=publication_state,
            locale=locale,
            start=0,
            batch_size=1,
            with_count=False,
        )
        return cls.first_from_list_response(response)

    @classmethod
    async def create_document(
        cls,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
        locale: str | None = None,
    ) -> Self:
        """Create a new document."""
        response = await client.create_document(
            plural_api_id=cls.__plural_api_id__,
            data=serialize_document_data(data),
            locale=locale,
        )
        _, populate = get_model_fields_and_population(cls)
        if not populate:
            return cls.from_scalar_response(response)
        else:
            result_document = BaseDocument.from_scalar_response(response)
            return await cls.get_document(client, result_document.document_id, locale=locale)

    def model_dump_data(self, exclude_managed_fields: bool = False, json_mode: bool = False) -> dict[str, Any]:
        """
        Create a dictionary representation of the document.

        Args:
            exclude_managed_fields: If True, exclude fields listed in __managed_fields__
            json_mode: If True, serialize fields to JSON compatible types

        Returns:
            Dictionary representation of the document with nested BaseDocument instances
            replaced with their IDs
        """
        return get_model_data(self, exclude_managed_fields=exclude_managed_fields, json_mode=json_mode)

    def model_identical(self, data: dict[str, Any] | BaseModel, exclude_fields: list[str] | None = None) -> bool:
        data_dict = data.model_dump(by_alias=True) if isinstance(data, BaseModel) else data
        data_dict_filtered = {k: v for k, v in data_dict.items() if k not in (exclude_fields or [])}
        record_dict = self.model_dump_data(exclude_managed_fields=True)
        record_dict_filtered = {k: v for k, v in record_dict.items() if k in data_dict_filtered}
        return hash_model(record_dict_filtered) == hash_model(data_dict_filtered)

    async def update_document(
        self,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
        lazy_mode: bool = False,
        do_not_compare_fields: list[str] | None = None,
    ) -> Self:
        """Update existing document."""
        _, populate = get_model_fields_and_population(self.__class__)
        if not lazy_mode and do_not_compare_fields:
            warnings.warn("do_not_compare_fields argument works only in lazy mode")
        elif lazy_mode and self.model_identical(data=data, exclude_fields=do_not_compare_fields):
            return self
        response = await client.update_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            data=serialize_document_data(data),
            locale=getattr(self, "locale", None),
        )
        if not populate:
            result_document = self.__class__.from_scalar_response(response)
            self.__dict__.update(result_document.__dict__)
            return self
        else:
            return await self.refresh_document(client)

    async def lazy_update_document(
        self,
        client: StrapiClientAsync,
        data: dict[str, Any] | BaseModel,
        do_not_compare_fields: list[str] | None = None,
    ) -> bool:
        """Lazy update existing document fields without record synchronization."""
        if self.model_identical(data=data, exclude_fields=do_not_compare_fields):
            return False
        else:
            await client.update_document(
                plural_api_id=self.__plural_api_id__,
                document_id=self.document_id,
                data=serialize_document_data(data),
                locale=getattr(self, "locale", None),
            )
            return True

    async def update_relations(
        self,
        client: StrapiClientAsync,
        field: str,
        relations: list[BaseDocument] | None = None,
        connect: list[BaseDocument] | None = None,
        disconnect: list[BaseDocument] | None = None,
    ) -> Self:
        """Update field relations."""
        if not relations and not connect and not disconnect:
            raise ValueError("At least one of relations, connect or disconnect should be provided")
        if relations and (connect or disconnect):
            raise ValueError("relations argument does not work with connect or disconnect arguments")
        if relations:
            data = {field: {"set": [d.document_id for d in relations]}}
        else:
            data = {
                field: {
                    "connect": [d.document_id for d in connect or {}],
                    "disconnect": [d.document_id for d in disconnect or {}],
                }
            }
        await client.update_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            data=data,
            locale=getattr(self, "locale", None),
        )
        return await self.refresh_document(client)

    async def refresh_document(self, client: StrapiClientAsync) -> Self:
        """Refresh the document with the latest data from Strapi."""
        fields, populate = get_model_fields_and_population(self.__class__)
        response = await client.get_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            fields=fields,
            populate=populate,
            locale=getattr(self, "locale", None),
        )
        document = self.from_scalar_response(response)
        self.__dict__.update(document.__dict__)
        return self

    async def delete_document(self, client: StrapiClientAsync) -> None:
        """Delete the document."""
        await client.delete_document(
            plural_api_id=self.__plural_api_id__,
            document_id=self.document_id,
            locale=getattr(self, "locale", None),
        )

    async def upload_file(
        self,
        client: StrapiClientAsync,
        file: Path | str | dict[str, BytesIO | bytes | bytearray | memoryview],
        field: str,
    ) -> Self:
        """Upload a file to the document's field."""
        if isinstance(file, dict):
            file_data: list[Path | str] | dict[str, BytesIO | bytes | bytearray | memoryview] = file
        else:
            file_data = [file]
        await client.upload_files(
            files=file_data,
            content_type_id=self.__content_type_id__,
            document_id=self.id,
            field=field,
        )
        return await self.refresh_document(client)