diff --git a/HISTORY.md b/HISTORY.md
index 6a57dc4..0a35b55 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -6,7 +6,7 @@ The format is based on [Keep a
Changelog](https://keepachangelog.com/en/1.0.0/), and this project
adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
-## [6.3.0](https://github.com/uploadcare/pyuploadcare/compare/v6.2.1...v6.3.0) - 2026-08-03
+## [6.3.0](https://github.com/uploadcare/pyuploadcare/compare/v6.2.1...v6.3.0) - 2026-08-04
### Added
@@ -22,6 +22,21 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
option for `ucare upload`.
- `TagValidationError`, raised when tags exceed the API limits (50 tags per file, 100 characters
per tag, Latin letters, digits, `-`, `_` and `.` only).
+- Support for [file search](https://uploadcare.com/docs/file-search/):
+ - `Uploadcare.search_files()`, returning one page of results, and
+ `Uploadcare.iterate_search_files()`, which walks pages. As with the other list APIs, `limit`
+ is the total number of results to yield and `request_limit` is the page size.
+ - `FilesAPI.search()` for `POST /files/search/`.
+ - Typed requests: `FileSearchRequest` with `query`, `phrase`, `exact`, `datetime_uploaded`,
+ `size`, `is_image` and `tags` conditions plus the `fuzziness` and `sort` modifiers, built from
+ `SearchPhrase`, `SearchExact`, `DatetimeRange`, `SizeRange`, `TagsFilter` and `SearchSort`.
+ A plain dict in the same shape is accepted too. Requests are validated locally against the
+ documented API constraints before a request is made.
+ - `FileSearchResponse` with `next`, `previous`, `total`, `per_page` and `results`, where each
+ result is a `FileSearchInfo` — file info plus a `SearchHighlight`.
+ - `iterate_search_files()` warns when asked to page through a filter-only request without
+ `sort`, whose result order the API leaves undefined.
+ - A new `ucare search_files` command.
### Changed
diff --git a/README.md b/README.md
index 9779f73..ea84fcf 100644
--- a/README.md
+++ b/README.md
@@ -33,6 +33,7 @@ Build file handling in minutes. Upload or accept user-generated content, store,
- [Usage](#usage)
- [Basic usage](#basic-usage)
- [File tags](#file-tags)
+ - [File search](#file-search)
- [Django integration](#django-integration)
- [Testing](#testing)
- [Demo app](#demo-app)
@@ -195,6 +196,64 @@ with open("sample-file.jpeg", "rb") as file_object:
Tags are lowercased, trimmed and deduplicated. A file can hold up to 50 tags of up to 100
characters each, made of Latin letters, digits, `-`, `_` and `.`.
+#### File search
+
+A single [search](https://uploadcare.com/docs/file-search/) request can combine full-text search,
+exact matching, range filters and tag filters. At least one condition is required:
+
+```python
+from pyuploadcare import FileSearchRequest, SizeRange, TagsFilter
+
+response = uploadcare.search_files(
+ FileSearchRequest(
+ query="sunset",
+ tags=TagsFilter(all_=["cat"], none_=["draft"]),
+ size=SizeRange(gt=1024),
+ is_image=True,
+ fuzziness=True,
+ sort=["-score", "size"],
+ ),
+ limit=50,
+)
+
+print(response.total, response.per_page)
+
+for file_info in response.results:
+ print(file_info.original_filename, file_info.tags)
+ if file_info.highlight:
+ print(file_info.highlight.original_filename) # ['sunset.jpg']
+```
+
+Requests can also be plain dicts:
+
+```python
+response = uploadcare.search_files({"tags": {"all": ["cat"]}})
+```
+
+`search_files()` returns a single page: `limit` is the page size (1–100, defaults to 20) and
+`offset + limit` must not exceed 1000.
+
+To walk pages, use `iterate_search_files()`. There, following the SDK's other list APIs, `limit` is
+the total number of results to yield and `request_limit` is the page size; the iterator stops on
+its own once it reaches the 1000-result window:
+
+```python
+request = {"tags": {"all": ["cat"]}, "sort": ["-datetime_uploaded"]}
+
+for file_info in uploadcare.iterate_search_files(request, limit=200):
+ print(file_info.uuid)
+```
+
+Always pass an explicit `sort` when paging through a filter-only request (one without `query` or
+`phrase`): there is no relevance to rank by, so the order is undefined and paging can skip or
+repeat files. The SDK emits a `UserWarning` if you don't.
+
+Either way, search reaches the first 1000 results only. Narrow the query rather than paging
+deeper.
+
+> `highlight` values contain your users' filenames and metadata wrapped in `` tags by the
+> server. Treat them as untrusted text and escape them before rendering as HTML.
+
### Django integration
Let's add [File Uploader](https://uploadcare.com/docs/file-uploader/) to an existing Django project.
diff --git a/docs/core_api.rst b/docs/core_api.rst
index 3a6d70c..ae2f68c 100644
--- a/docs/core_api.rst
+++ b/docs/core_api.rst
@@ -268,6 +268,125 @@ Tags are available from the CLI as well::
.. _file tags documentation: https://uploadcare.com/docs/file-tags/
+Searching files
+---------------
+
+A single search request (`file search documentation`_) can combine full-text search, exact
+matching, range filters and tag filters. At least one condition is required: ``query``,
+``phrase``, ``exact``, ``datetime_uploaded``, ``size``, ``is_image`` or ``tags``. ``fuzziness``
+and ``sort`` are modifiers and do not count as conditions::
+
+ from pyuploadcare import (
+ DatetimeRange,
+ FileSearchRequest,
+ SearchExact,
+ SearchPhrase,
+ SizeRange,
+ TagsFilter,
+ )
+
+ response = uploadcare.search_files(
+ FileSearchRequest(
+ query='sunset',
+ tags=TagsFilter(all_=['cat'], none_=['draft']),
+ size=SizeRange(gt=1024, lte=10 * 1024 * 1024),
+ is_image=True,
+ fuzziness=True,
+ sort=['-score', 'size'],
+ ),
+ limit=50,
+ )
+
+ print(response.total, response.per_page, response.next)
+
+ for file_info in response.results:
+ print(file_info.original_filename, file_info.tags)
+
+``query`` searches across all searchable fields and must be at least 4 characters. ``phrase``
+performs an ordered full-text match on a specific field, and ``exact`` matches values exactly::
+
+ request = FileSearchRequest(
+ phrase=SearchPhrase(original_filename='holiday photo'),
+ exact=SearchExact(detected_mime_type=['image/jpeg', 'image/png']),
+ datetime_uploaded=DatetimeRange(gte=datetime(2024, 1, 1)),
+ )
+
+A field cannot appear in both ``phrase`` and ``exact`` in the same request.
+
+``exact`` can also match metadata values. In the SDK this is a nested mapping, which is
+serialized into the ``metadata[]`` keys the API expects::
+
+ SearchExact(metadata={'album': ['holiday'], 'color': ['red']})
+
+A request can equally be a plain dict, using the same shape::
+
+ response = uploadcare.search_files({
+ 'exact': {'metadata': {'album': ['holiday']}},
+ 'tags': {'all': ['cat']},
+ })
+
+``sort`` accepts 1 to 4 keys out of ``score``, ``datetime_uploaded``, ``size`` and
+``original_filename``, each optionally prefixed with ``-`` for descending order. Keys must be
+unique and the same key must not be given in both directions.
+
+Without ``sort``, results are ordered by relevance. A filter-only request has no ``query`` or
+``phrase`` to rank by, so its order is undefined — always give such a request an explicit
+``sort``.
+
+Search indexing is asynchronous, so a file is not findable the instant it is uploaded — expect a
+delay on the order of seconds. Do not search for a file you have just uploaded without retrying.
+
+``search_files`` returns a single page. Its ``limit`` is the page size, 1 to 100, and the server
+defaults to 20; ``offset + limit`` must not exceed 1000. Search cannot reach past the first 1000
+results, so narrow the query instead of paging deeper.
+
+To walk pages, use ``iterate_search_files``. Its ``limit`` means something different: as elsewhere
+in the SDK, ``limit`` is the total number of results to yield and ``request_limit`` is the number
+retrieved per request. The iterator applies the 1000-result window itself, clamping each page so
+that no request exceeds it::
+
+ for file_info in uploadcare.iterate_search_files(
+ {'tags': {'all': ['cat']}, 'sort': ['-datetime_uploaded']}, limit=200
+ ):
+ print(file_info.uuid)
+
+.. warning::
+
+ Paging through a filter-only request without ``sort`` is unreliable: the order is undefined
+ between requests, so files may be skipped or repeated. ``iterate_search_files`` emits a
+ ``UserWarning`` in that case.
+
+Pass ``include_appdata=True`` to embed application data in every result::
+
+ response = uploadcare.search_files(request, include_appdata=True)
+ print(response.results[0].appdata)
+
+Each result carries a ``highlight`` with the matched tokens wrapped in ```` tags. A field in it
+is populated only when that field matched a full-text condition, so a filter-only search
+highlights nothing — expect either no ``highlight`` at all or one whose every field is ``None``::
+
+ highlight = response.results[0].highlight
+ if highlight:
+ print(highlight.original_filename) # ['sunset-cat.jpg']
+ print(highlight.metadata) # {'album': 'summer sunset'}
+
+.. warning::
+
+ ``highlight`` values contain user-controlled content (filenames, metadata) plus markup added
+ by the server. They are not trusted HTML — escape them before rendering.
+
+Search is available from the CLI as well. Because a descending sort key starts with a dash, pass
+it with ``--sort=`` so it is not read as another option::
+
+ ucare search_files --query sunset --tags_any cat dog --is_image true \
+ --size_gt 1024 --sort=-score --limit 50
+
+The CLI covers every condition except ``exact.metadata``, whose bracketed keys do not map onto a
+command line option; use the Python API for that.
+
+.. _file search documentation: https://uploadcare.com/docs/file-search/
+
+
Video conversion
----------------
diff --git a/pyuploadcare/__init__.py b/pyuploadcare/__init__.py
index ef83bee..a44f6d3 100644
--- a/pyuploadcare/__init__.py
+++ b/pyuploadcare/__init__.py
@@ -6,4 +6,13 @@
from pyuploadcare.resources.file_list import FileList # noqa: F401
from pyuploadcare.resources.group_list import GroupList # noqa: F401
from pyuploadcare.api.entities import Webhook, ProjectInfo # noqa: F401
+from pyuploadcare.api.search_entities import ( # noqa: F401
+ DatetimeRange,
+ FileSearchRequest,
+ SearchExact,
+ SearchPhrase,
+ SearchSort,
+ SizeRange,
+ TagsFilter,
+)
from pyuploadcare.client import Uploadcare # noqa: F401
diff --git a/pyuploadcare/api/api.py b/pyuploadcare/api/api.py
index 4c46065..4258f0c 100644
--- a/pyuploadcare/api/api.py
+++ b/pyuploadcare/api/api.py
@@ -35,13 +35,21 @@
from .entities import UUIDEntity
from .metadata import validate_meta_key, validate_meta_value, validate_metadata
+from .search_entities import FileSearchRequest
from .tags import validate_tags
-from .utils import flatten_dict
+from .utils import flatten_dict, require_optional_int, require_range
logger = logging.getLogger("pyuploadcare")
+# File search pagination limits.
+# https://uploadcare.com/docs/api/rest/file/search-files/
+SEARCH_DEFAULT_LIMIT = 20 # the server default when `limit` is not sent
+SEARCH_MAX_LIMIT = 100
+SEARCH_MAX_WINDOW = 1000 # `offset` + `limit` must not exceed this
+
+
class FilesAPI(API, ListCountMixin, RetrieveMixin, DeleteWithResponseMixin):
resource_type = "files"
response_classes = {
@@ -55,8 +63,70 @@ class FilesAPI(API, ListCountMixin, RetrieveMixin, DeleteWithResponseMixin):
"batch_delete": responses.BatchFileOperationResponse,
"local_copy": responses.CreateLocalCopyResponse,
"remote_copy": responses.CreateRemoteCopyResponse,
+ "search": responses.FileSearchResponse,
}
+ def search(
+ self,
+ request: Union[FileSearchRequest, Dict[str, Any]],
+ limit: Optional[int] = None,
+ offset: Optional[int] = None,
+ include_appdata: bool = False,
+ ) -> responses.FileSearchResponse:
+ """Search files, returning a single page of results.
+
+ https://uploadcare.com/docs/file-search/
+
+ Args:
+ - request: a ``FileSearchRequest`` or a dict in the same shape.
+ A dict uses the SDK's shape for ``exact.metadata``, i.e.
+ ``{"exact": {"metadata": {"color": ["red"]}}}``, not the wire
+ shape ``{"exact": {"metadata[color]": [...]}}``.
+ - limit: results per page, 1 to 100. The server defaults to 20.
+ - offset: how many results to skip. ``offset + limit`` must not
+ exceed 1000.
+ - include_appdata: embed application data in every result.
+ """
+ search_request = (
+ request
+ if isinstance(request, FileSearchRequest)
+ else FileSearchRequest.model_validate(request)
+ )
+
+ require_optional_int("limit", limit)
+ require_optional_int("offset", offset)
+ require_range("limit", limit, minimum=1, maximum=SEARCH_MAX_LIMIT)
+ require_range("offset", offset, minimum=0)
+
+ effective_limit = SEARCH_DEFAULT_LIMIT if limit is None else limit
+ effective_offset = 0 if offset is None else offset
+
+ if effective_offset + effective_limit > SEARCH_MAX_WINDOW:
+ raise InvalidParamError(
+ "`offset` + `limit` must not exceed "
+ f"{SEARCH_MAX_WINDOW}, got "
+ f"{effective_offset} + {effective_limit}. "
+ "Narrow the query instead of paging deeper"
+ )
+
+ query_parameters: Dict[str, Any] = {}
+ if limit is not None:
+ query_parameters["limit"] = limit
+ if offset is not None:
+ query_parameters["offset"] = offset
+ if include_appdata:
+ query_parameters["include"] = "appdata"
+
+ url = self._build_url(
+ suffix="search", query_parameters=query_parameters
+ )
+ response_class = self._get_response_class("search")
+ json_response = self._client.post(
+ url, json=search_request.to_payload()
+ ).json()
+ response = self._parse_response(json_response, response_class)
+ return cast(responses.FileSearchResponse, response)
+
def store(self, file_uuid: Union[UUID, str]) -> entities.FileInfo:
url = self._build_url(file_uuid, suffix="storage")
response_class = self._get_response_class("store")
diff --git a/pyuploadcare/api/entities.py b/pyuploadcare/api/entities.py
index caaa391..60799b2 100644
--- a/pyuploadcare/api/entities.py
+++ b/pyuploadcare/api/entities.py
@@ -240,6 +240,33 @@ class FileInfo(UUIDEntity):
appdata: Optional[ApplicationDataSet] = None
+class SearchHighlight(Entity):
+ """Matched tokens wrapped in ```` tags by the search backend.
+
+ A field is populated only when it matched a full-text condition
+ (``query`` or ``phrase``). Without such a condition nothing is
+ highlighted: the API may then omit the object altogether, or send it
+ empty, in which case every field here is ``None``.
+
+ The values contain user-controlled content (filenames, metadata) plus
+ markup added by the server. They are **not** trusted HTML: escape the
+ surrounding text before rendering them.
+ """
+
+ # OpenAPI: array of string
+ original_filename: Optional[List[str]] = None
+ # OpenAPI: array of string
+ detected_mime_type: Optional[List[str]] = None
+ # OpenAPI: object with additionalProperties of type string
+ metadata: Optional[Dict[str, str]] = None
+
+
+class FileSearchInfo(FileInfo):
+ """A file search result: file info plus the match highlight."""
+
+ highlight: Optional[SearchHighlight] = None
+
+
class GroupInfo(Entity):
id: str
_fetched: Optional[bool] = PrivateAttr(default=False)
diff --git a/pyuploadcare/api/responses.py b/pyuploadcare/api/responses.py
index 9333ad6..3f4aff1 100644
--- a/pyuploadcare/api/responses.py
+++ b/pyuploadcare/api/responses.py
@@ -8,6 +8,7 @@
DocumentConvertInfo,
Entity,
FileInfo,
+ FileSearchInfo,
GroupInfo,
MetadataDict,
VideoConvertInfo,
@@ -33,6 +34,11 @@ class FileListResponse(PaginatedResponse):
results: List[FileInfo] # type: ignore
+class FileSearchResponse(PaginatedResponse):
+ # https://uploadcare.com/docs/api/rest/file/search-files/
+ results: List[FileSearchInfo] # type: ignore
+
+
class GroupListResponse(PaginatedResponse):
# https://uploadcare.com/api-refs/rest-api/v0.5.0/#operation/groupsList
results: List[GroupInfo] # type: ignore
diff --git a/pyuploadcare/api/search_entities.py b/pyuploadcare/api/search_entities.py
new file mode 100644
index 0000000..40a3091
--- /dev/null
+++ b/pyuploadcare/api/search_entities.py
@@ -0,0 +1,292 @@
+"""Request models for file search.
+
+https://uploadcare.com/docs/file-search/
+https://uploadcare.com/docs/api/rest/file/search-files/
+"""
+
+from datetime import datetime
+from enum import Enum
+from typing import Any, ClassVar, Dict, List, Optional, Tuple
+
+from pydantic import (
+ BaseModel,
+ ConfigDict,
+ Field,
+ StrictBool,
+ StrictInt,
+ model_validator,
+)
+
+from .metadata import validate_meta_key
+from .tags import validate_tags
+
+
+# The shortest accepted full-text term, per the API reference.
+MIN_TERM_LENGTH = 4
+
+# `sort` accepts 1-4 unique keys.
+MIN_SORT_KEYS = 1
+MAX_SORT_KEYS = 4
+
+
+class SearchSort(str, Enum):
+ """Sort keys accepted by file search. `-` prefix means descending."""
+
+ SCORE = "score"
+ SCORE_DESC = "-score"
+ DATETIME_UPLOADED = "datetime_uploaded"
+ DATETIME_UPLOADED_DESC = "-datetime_uploaded"
+ SIZE = "size"
+ SIZE_DESC = "-size"
+ ORIGINAL_FILENAME = "original_filename"
+ ORIGINAL_FILENAME_DESC = "-original_filename"
+
+
+class SearchRequestModel(BaseModel):
+ """Base for search request models.
+
+ ``extra="forbid"`` so a misspelled key is reported instead of silently
+ dropped, which matters most when a request is built from a plain dict.
+ """
+
+ model_config = ConfigDict(extra="forbid", populate_by_name=True)
+
+ def _set_field_names(self) -> List[str]:
+ """Names of the fields that carry an actual condition."""
+ return [
+ name for name, value in self.__dict__.items() if value is not None
+ ]
+
+
+class DatetimeRange(SearchRequestModel):
+ gt: Optional[datetime] = None
+ gte: Optional[datetime] = None
+ lt: Optional[datetime] = None
+ lte: Optional[datetime] = None
+
+ @model_validator(mode="after")
+ def _at_least_one_bound(self) -> "DatetimeRange":
+ if not self._set_field_names():
+ raise ValueError(
+ "at least one of `gt`, `gte`, `lt` or `lte` is required"
+ )
+ return self
+
+
+class SizeRange(SearchRequestModel):
+ # Strict, so `gt=True` is rejected instead of silently becoming `gt=1`.
+ gt: Optional[StrictInt] = Field(None, ge=0)
+ gte: Optional[StrictInt] = Field(None, ge=0)
+ lt: Optional[StrictInt] = Field(None, ge=0)
+ lte: Optional[StrictInt] = Field(None, ge=0)
+
+ @model_validator(mode="after")
+ def _at_least_one_bound(self) -> "SizeRange":
+ if not self._set_field_names():
+ raise ValueError(
+ "at least one of `gt`, `gte`, `lt` or `lte` is required"
+ )
+ return self
+
+
+class SearchPhrase(SearchRequestModel):
+ """Ordered full-text match. Each value must be at least 4 characters."""
+
+ original_filename: Optional[str] = Field(None, min_length=MIN_TERM_LENGTH)
+ metadata: Optional[str] = Field(None, min_length=MIN_TERM_LENGTH)
+ detected_mime_type: Optional[str] = Field(None, min_length=MIN_TERM_LENGTH)
+
+ @model_validator(mode="after")
+ def _at_least_one_field(self) -> "SearchPhrase":
+ if not self._set_field_names():
+ raise ValueError("`phrase` requires at least one field")
+ return self
+
+
+class SearchExact(SearchRequestModel):
+ """Exact matching. Each key takes a non-empty array of values."""
+
+ uuid: Optional[List[str]] = Field(None, min_length=1)
+ detected_mime_type: Optional[List[str]] = Field(None, min_length=1)
+ original_filename: Optional[List[str]] = Field(None, min_length=1)
+ # Serialized as `metadata[]` by `FileSearchRequest.to_payload()`.
+ metadata: Optional[Dict[str, List[str]]] = None
+
+ @model_validator(mode="after")
+ def _validate_metadata(self) -> "SearchExact":
+ if self.metadata is None:
+ return self
+
+ for key, values in self.metadata.items():
+ # Keeps a `[` or `]` in a key from forging a different wire key.
+ validate_meta_key(key)
+
+ if not values:
+ raise ValueError(
+ f"`exact.metadata[{key}]` requires a non-empty array"
+ )
+
+ return self
+
+ @model_validator(mode="after")
+ def _at_least_one_condition(self) -> "SearchExact":
+ # An empty `metadata` mapping carries no condition, so it must not
+ # count towards this check: it would serialize to `"exact": {}`.
+ has_condition = (
+ self.uuid is not None
+ or self.detected_mime_type is not None
+ or self.original_filename is not None
+ or bool(self.metadata)
+ )
+
+ if not has_condition:
+ raise ValueError("`exact` requires at least one condition")
+
+ return self
+
+
+class TagsFilter(SearchRequestModel):
+ """Tag filters.
+
+ ``any``, ``all`` and ``none`` shadow Python builtins and keywords, so the
+ attributes are suffixed with an underscore and aliased to the wire names.
+ Both ``TagsFilter(any_=[...])`` and ``TagsFilter(**{"any": [...]})`` work.
+ """
+
+ any_: Optional[List[str]] = Field(None, alias="any")
+ all_: Optional[List[str]] = Field(None, alias="all")
+ none_: Optional[List[str]] = Field(None, alias="none")
+
+ @model_validator(mode="after")
+ def _normalize_and_check(self) -> "TagsFilter":
+ for name in ("any_", "all_", "none_"):
+ value = getattr(self, name)
+ if value is not None:
+ # Stored tags are normalized, so filters have to be too.
+ # No count limit: the 50-tag ceiling applies to the tags
+ # stored on a single file, not to a filter's alternatives.
+ object.__setattr__(
+ self, name, validate_tags(value, max_count=None)
+ )
+
+ if not any(getattr(self, name) for name in ("any_", "all_", "none_")):
+ raise ValueError(
+ "`tags` requires at least one non-empty list of tags"
+ )
+
+ return self
+
+
+class FileSearchRequest(SearchRequestModel):
+ """A file search request.
+
+ At least one condition is required: ``query``, ``phrase``, ``exact``,
+ ``datetime_uploaded``, ``size``, ``is_image`` or ``tags``. ``fuzziness``
+ and ``sort`` are modifiers and do not count as conditions.
+ """
+
+ query: Optional[str] = Field(None, min_length=MIN_TERM_LENGTH)
+ phrase: Optional[SearchPhrase] = None
+ exact: Optional[SearchExact] = None
+ datetime_uploaded: Optional[DatetimeRange] = None
+ size: Optional[SizeRange] = None
+ # Strict, so `1` or `"true"` is rejected rather than silently coerced.
+ is_image: Optional[StrictBool] = None
+ tags: Optional[TagsFilter] = None
+ fuzziness: Optional[StrictBool] = None
+ sort: Optional[List[SearchSort]] = Field(
+ None, min_length=MIN_SORT_KEYS, max_length=MAX_SORT_KEYS
+ )
+
+ CONDITION_FIELDS: ClassVar[Tuple[str, ...]] = (
+ "query",
+ "phrase",
+ "exact",
+ "datetime_uploaded",
+ "size",
+ "is_image",
+ "tags",
+ )
+
+ # Fields `phrase` and `exact` have in common. `metadata` is excluded on
+ # purpose: `phrase.metadata` and `exact.metadata[]` are different
+ # field names on the wire, so they do not collide.
+ OVERLAPPING_FIELDS: ClassVar[Tuple[str, ...]] = (
+ "original_filename",
+ "detected_mime_type",
+ )
+
+ @model_validator(mode="after")
+ def _at_least_one_condition(self) -> "FileSearchRequest":
+ if all(getattr(self, name) is None for name in self.CONDITION_FIELDS):
+ raise ValueError(
+ "at least one of "
+ + ", ".join(f"`{name}`" for name in self.CONDITION_FIELDS)
+ + " is required"
+ )
+ return self
+
+ @model_validator(mode="after")
+ def _no_phrase_and_exact_overlap(self) -> "FileSearchRequest":
+ if self.phrase is None or self.exact is None:
+ return self
+
+ for name in self.OVERLAPPING_FIELDS:
+ if (
+ getattr(self.phrase, name) is not None
+ and getattr(self.exact, name) is not None
+ ):
+ raise ValueError(
+ f"`{name}` cannot appear in both `phrase` and `exact`"
+ )
+
+ return self
+
+ @model_validator(mode="after")
+ def _unique_sort_keys(self) -> "FileSearchRequest":
+ if self.sort is None:
+ return self
+
+ seen = set()
+
+ for key in self.sort:
+ # Both directions of one key count as the same key.
+ field = key.value.lstrip("-")
+
+ if field in seen:
+ raise ValueError(
+ f"`sort` must not contain `{field}` more than once, "
+ "in either direction"
+ )
+
+ seen.add(field)
+
+ return self
+
+ def has_undefined_order(self) -> bool:
+ """Whether the result order of this request is undefined.
+
+ Without ``sort``, results come back ranked by relevance. A filter-only
+ request has no ``query`` or ``phrase`` to rank by, so its order is
+ undefined and an explicit ``sort`` is required to page reliably.
+ """
+ return self.sort is None and self.query is None and self.phrase is None
+
+ def to_payload(self) -> Dict[str, Any]:
+ """Render the request as the JSON body the API expects."""
+ payload = self.model_dump(
+ mode="json", by_alias=True, exclude_none=True
+ )
+
+ exact = payload.get("exact")
+ if exact and "metadata" in exact:
+ for key, values in exact.pop("metadata").items():
+ exact[f"metadata[{key}]"] = values
+
+ tags = payload.get("tags")
+ if tags:
+ payload["tags"] = {
+ key: value for key, value in tags.items() if value
+ }
+
+ return payload
diff --git a/pyuploadcare/api/utils.py b/pyuploadcare/api/utils.py
index 8318831..a960214 100644
--- a/pyuploadcare/api/utils.py
+++ b/pyuploadcare/api/utils.py
@@ -1,3 +1,41 @@
+from typing import Any, Optional
+
+from pyuploadcare.exceptions import InvalidParamError
+
+
+def require_optional_int(name: str, value: Any) -> None:
+ """Reject a value that is neither ``None`` nor a real int.
+
+ Type annotations are not enforced at runtime, and ``bool`` is a subclass
+ of ``int``, so ``limit=True`` would otherwise reach the query string
+ verbatim as ``limit=True``.
+ """
+ if value is None:
+ return
+
+ if isinstance(value, bool) or not isinstance(value, int):
+ raise InvalidParamError(
+ f"`{name}` must be an int, got {type(value).__name__}"
+ )
+
+
+def require_range(
+ name: str,
+ value: Optional[int],
+ minimum: Optional[int] = None,
+ maximum: Optional[int] = None,
+) -> None:
+ """Reject an out-of-range value. ``None`` always passes."""
+ if value is None:
+ return
+
+ if minimum is not None and value < minimum:
+ raise InvalidParamError(f"`{name}` must be >= {minimum}, got {value}")
+
+ if maximum is not None and value > maximum:
+ raise InvalidParamError(f"`{name}` must be <= {maximum}, got {value}")
+
+
def flatten_dict(simple_mapping, attribute_base="metadata") -> dict:
"""
Straightforward way to use nested dict for multipart/form-data
diff --git a/pyuploadcare/client.py b/pyuploadcare/client.py
index e2d096a..5c67c7b 100644
--- a/pyuploadcare/client.py
+++ b/pyuploadcare/client.py
@@ -1,6 +1,7 @@
import os
import socket
import ssl
+import warnings
from time import time
from typing import (
IO,
@@ -8,6 +9,7 @@
Callable,
Dict,
Iterable,
+ Iterator,
List,
Optional,
Tuple,
@@ -29,10 +31,23 @@
VideoConvertAPI,
WebhooksAPI,
)
-from pyuploadcare.api.api import URLAPI
+from pyuploadcare.api.api import (
+ SEARCH_DEFAULT_LIMIT,
+ SEARCH_MAX_LIMIT,
+ SEARCH_MAX_WINDOW,
+ URLAPI,
+)
from pyuploadcare.api.auth import UploadcareAuth
from pyuploadcare.api.client import Client
-from pyuploadcare.api.entities import ProjectInfo, Webhook, WebhookEvent
+from pyuploadcare.api.entities import (
+ FileSearchInfo,
+ ProjectInfo,
+ Webhook,
+ WebhookEvent,
+)
+from pyuploadcare.api.responses import FileSearchResponse
+from pyuploadcare.api.search_entities import FileSearchRequest
+from pyuploadcare.api.utils import require_optional_int, require_range
from pyuploadcare.exceptions import DuplicateFileError, InvalidParamError
from pyuploadcare.helpers import (
get_file_size,
@@ -783,6 +798,202 @@ def list_files(
removed=removed,
)
+ def search_files(
+ self,
+ request: Union[FileSearchRequest, Dict[str, Any]],
+ limit: Optional[int] = None,
+ offset: Optional[int] = None,
+ include_appdata: bool = False,
+ ) -> FileSearchResponse:
+ """Search files, returning a single page of results.
+
+ One request can combine full-text search, exact matching, range
+ filters and tag filters. At least one condition is required.
+
+ Usage example::
+
+ >>> from pyuploadcare import FileSearchRequest, TagsFilter
+ >>> response = uploadcare.search_files(
+ ... FileSearchRequest(
+ ... query='sunset',
+ ... tags=TagsFilter(all_=['cat']),
+ ... sort=['-score'],
+ ... ),
+ ... limit=50,
+ ... )
+ >>> response.total
+ 2
+ >>> for file_info in response.results:
+ ... print(file_info.original_filename, file_info.tags)
+
+ The response is returned as is, because ``total``, ``per_page``,
+ ``next``, ``previous`` and the per-result ``highlight`` are all
+ meaningful. Use ``iterate_search_files`` to walk pages instead.
+
+ Args:
+ - request: a ``FileSearchRequest`` or a dict in the same shape.
+ - limit: results per page, 1 to 100. The server defaults to 20.
+ - offset: how many results to skip. ``offset + limit`` must not
+ exceed 1000.
+ - include_appdata: embed application data in every result.
+
+ Returns:
+ ``FileSearchResponse``
+
+ """
+ return self.files_api.search(
+ request,
+ limit=limit,
+ offset=offset,
+ include_appdata=include_appdata,
+ )
+
+ def iterate_search_files(
+ self,
+ request: Union[FileSearchRequest, Dict[str, Any]],
+ limit: Optional[int] = None,
+ request_limit: Optional[int] = None,
+ offset: Optional[int] = None,
+ include_appdata: bool = False,
+ ) -> Iterator[FileSearchInfo]:
+ """Iterate over file search results, page by page.
+
+ Usage example::
+
+ >>> for file_info in uploadcare.iterate_search_files(
+ ... {'tags': {'all': ['cat']}, 'sort': ['-datetime_uploaded']},
+ ... limit=200,
+ ... ):
+ ... print(file_info.uuid)
+
+ Always pass an explicit ``sort`` for a filter-only request, i.e. one
+ without ``query`` or ``phrase``. Such a request has no relevance to
+ rank by, so its order is undefined, and paging through an undefined
+ order can skip or repeat files. A ``UserWarning`` is emitted when that
+ happens.
+
+ Search is limited to the first 1000 results; narrow the query instead
+ of paging deeper.
+
+ Args:
+ - request: a ``FileSearchRequest`` or a dict in the same shape.
+ - limit: total number of results to yield. ``None`` yields
+ everything reachable.
+ - request_limit: number of results retrieved per request (page).
+ Usually, you don't need worry about this parameter.
+ - offset: how many results to skip before the first page.
+ - include_appdata: embed application data in every result.
+
+ """
+ require_optional_int("limit", limit)
+ require_optional_int("request_limit", request_limit)
+ require_optional_int("offset", offset)
+ require_range("limit", limit, minimum=0)
+ require_range(
+ "request_limit", request_limit, minimum=1, maximum=SEARCH_MAX_LIMIT
+ )
+ require_range("offset", offset, minimum=0, maximum=SEARCH_MAX_WINDOW)
+
+ # Validate the request here rather than inside the generator, so an
+ # invalid request is reported immediately instead of on first
+ # iteration. It also keeps every page from re-validating it.
+ search_request = (
+ request
+ if isinstance(request, FileSearchRequest)
+ else FileSearchRequest.model_validate(request)
+ )
+
+ if search_request.has_undefined_order():
+ warnings.warn(
+ "Paging through a filter-only search without `sort` is "
+ "unreliable: the result order is undefined, so files may be "
+ "skipped or repeated. Pass an explicit `sort`.",
+ UserWarning,
+ stacklevel=2,
+ )
+
+ return self._iterate_search_files(
+ search_request,
+ limit=limit,
+ request_limit=request_limit,
+ offset=offset,
+ include_appdata=include_appdata,
+ )
+
+ @staticmethod
+ def _search_page_size(
+ page_size: int, current_offset: int, remaining: Optional[int]
+ ) -> int:
+ """Page size for the next search request, or 0 when done.
+
+ Clamped to the search window, because a legal starting offset can
+ otherwise produce an illegal ``offset`` + ``limit`` combination, and to
+ what is left of the caller's total limit.
+ """
+ size = min(page_size, SEARCH_MAX_WINDOW - current_offset)
+
+ if remaining is not None:
+ size = min(size, remaining)
+
+ return max(size, 0)
+
+ def _iterate_search_files( # noqa: C901
+ self,
+ request: FileSearchRequest,
+ limit: Optional[int],
+ request_limit: Optional[int],
+ offset: Optional[int],
+ include_appdata: bool,
+ ) -> Iterator[FileSearchInfo]:
+ """Walk search result pages.
+
+ The response's ``next`` URL is deliberately not requested: it is an
+ absolute, server-supplied URL, and the REST client attaches
+ credentials to whatever URL it is given. So ``next`` is used only as a
+ "there is more" signal and the next offset is computed locally.
+ """
+ page_size = (
+ SEARCH_DEFAULT_LIMIT if request_limit is None else request_limit
+ )
+ current_offset = 0 if offset is None else offset
+ remaining = limit
+
+ while True:
+ current_page_size = self._search_page_size(
+ page_size, current_offset, remaining
+ )
+
+ if not current_page_size:
+ return
+
+ response = self.files_api.search(
+ request,
+ limit=current_page_size,
+ offset=current_offset,
+ include_appdata=include_appdata,
+ )
+
+ if not response.results:
+ return
+
+ for file_info in response.results:
+ yield file_info
+
+ if remaining is not None:
+ remaining -= 1
+ if remaining <= 0:
+ return
+
+ # Advance by the requested page size, not by how many results came
+ # back. This is offset pagination: a short page that still reports
+ # a `next` would otherwise make the following request overlap it.
+ current_offset += current_page_size
+
+ # `total` is not used as a stop condition: the API documents it as
+ # possibly approximate for very large result sets.
+ if response.next is None:
+ return
+
def list_file_groups(
self,
starting_point=None,
diff --git a/pyuploadcare/ucare_cli/commands/helpers.py b/pyuploadcare/ucare_cli/commands/helpers.py
index 61538e9..79133eb 100644
--- a/pyuploadcare/ucare_cli/commands/helpers.py
+++ b/pyuploadcare/ucare_cli/commands/helpers.py
@@ -1,3 +1,4 @@
+import argparse
import json
import sys
import time
@@ -88,6 +89,22 @@ def bool_or_none(value):
return {"true": True, "false": False}.get(value)
+def strict_bool(value):
+ """Parse a boolean, erroring out on anything unrecognised.
+
+ Unlike ``bool_or_none``, which silently maps unknown input to ``None``.
+ """
+ mapping = {"true": True, "false": False}
+ normalized = str(value).strip().lower()
+
+ if normalized not in mapping:
+ raise argparse.ArgumentTypeError(
+ f"invalid boolean value: {value!r} (expected true or false)"
+ )
+
+ return mapping[normalized]
+
+
def int_or_none(value):
return None if value.lower() == "none" else int(value)
diff --git a/pyuploadcare/ucare_cli/commands/search_files.py b/pyuploadcare/ucare_cli/commands/search_files.py
new file mode 100644
index 0000000..d30c02e
--- /dev/null
+++ b/pyuploadcare/ucare_cli/commands/search_files.py
@@ -0,0 +1,157 @@
+from dateutil import parser as datetime_parser
+from pydantic import ValidationError
+
+from pyuploadcare.api.search_entities import FileSearchRequest, SearchSort
+from pyuploadcare.client import Uploadcare
+from pyuploadcare.exceptions import InvalidParamError
+from pyuploadcare.ucare_cli.commands.helpers import pprint, strict_bool
+
+
+def register_arguments(subparsers): # noqa: C901
+ subparser = subparsers.add_parser("search_files", help="search files")
+ subparser.set_defaults(func=search_files)
+
+ subparser.add_argument(
+ "--query",
+ help="full text search across all searchable fields,"
+ " at least 4 characters",
+ )
+ subparser.add_argument(
+ "--fuzziness",
+ action="store_true",
+ default=None,
+ help="allow approximate matches in the full text search",
+ )
+
+ for field in (
+ "original_filename",
+ "metadata",
+ "detected_mime_type",
+ ):
+ subparser.add_argument(
+ f"--phrase_{field}",
+ help=f"ordered full text match on {field},"
+ " at least 4 characters",
+ )
+
+ for field in ("uuid", "original_filename", "detected_mime_type"):
+ subparser.add_argument(
+ f"--exact_{field}",
+ nargs="+",
+ metavar="VALUE",
+ help=f"exact match on {field}",
+ )
+
+ for bound in ("gt", "gte", "lt", "lte"):
+ subparser.add_argument(
+ f"--size_{bound}",
+ type=int,
+ metavar="BYTES",
+ help=f"file size {bound} filter, in bytes",
+ )
+ subparser.add_argument(
+ f"--uploaded_{bound}",
+ type=datetime_parser.parse,
+ metavar="DATETIME",
+ help=f"upload datetime {bound} filter",
+ )
+
+ subparser.add_argument(
+ "--is_image",
+ type=strict_bool,
+ metavar="true|false",
+ help="filter images",
+ )
+
+ for name in ("any", "all", "none"):
+ subparser.add_argument(
+ f"--tags_{name}",
+ nargs="+",
+ metavar="TAG",
+ help=f"match files having {name} of these tags",
+ )
+
+ subparser.add_argument(
+ "--sort",
+ action="append",
+ choices=[key.value for key in SearchSort],
+ help="sort key, repeat for up to 4 keys. Prefix with `-` for"
+ " descending order, using the `--sort=-score` form so that the"
+ " leading dash is not read as another option",
+ )
+ subparser.add_argument(
+ "--limit",
+ type=int,
+ help="results per page, 1 to 100. Defaults to 20",
+ )
+ subparser.add_argument(
+ "--offset",
+ type=int,
+ help="results to skip. `offset` + `limit` must not exceed 1000",
+ )
+ subparser.add_argument(
+ "--include_appdata",
+ action="store_true",
+ help="embed application data in every result",
+ )
+ return subparser
+
+
+def _sub_condition(arg_namespace, prefix, fields):
+ """Collect `--_` arguments into a nested dict."""
+ condition = {
+ field: getattr(arg_namespace, f"{prefix}_{field}", None)
+ for field in fields
+ }
+ condition = {
+ field: value for field, value in condition.items() if value is not None
+ }
+ return condition or None
+
+
+def _build_request(arg_namespace) -> dict:
+ request = {
+ "query": arg_namespace.query,
+ "fuzziness": arg_namespace.fuzziness,
+ "is_image": arg_namespace.is_image,
+ "sort": arg_namespace.sort,
+ "phrase": _sub_condition(
+ arg_namespace,
+ "phrase",
+ ("original_filename", "metadata", "detected_mime_type"),
+ ),
+ "exact": _sub_condition(
+ arg_namespace,
+ "exact",
+ ("uuid", "original_filename", "detected_mime_type"),
+ ),
+ "size": _sub_condition(
+ arg_namespace, "size", ("gt", "gte", "lt", "lte")
+ ),
+ "datetime_uploaded": _sub_condition(
+ arg_namespace, "uploaded", ("gt", "gte", "lt", "lte")
+ ),
+ "tags": _sub_condition(arg_namespace, "tags", ("any", "all", "none")),
+ }
+ return {
+ name: value for name, value in request.items() if value is not None
+ }
+
+
+def search_files(arg_namespace, client: Uploadcare):
+ try:
+ request = FileSearchRequest.model_validate(
+ _build_request(arg_namespace)
+ )
+ except ValidationError as error:
+ # `main()` only handles UploadcareException, so a raw ValidationError
+ # would surface as a traceback.
+ raise InvalidParamError(str(error))
+
+ response = client.search_files(
+ request,
+ limit=arg_namespace.limit,
+ offset=arg_namespace.offset,
+ include_appdata=arg_namespace.include_appdata,
+ )
+ pprint(response.model_dump())
diff --git a/pyuploadcare/ucare_cli/main.py b/pyuploadcare/ucare_cli/main.py
index fd75d82..8d16303 100644
--- a/pyuploadcare/ucare_cli/main.py
+++ b/pyuploadcare/ucare_cli/main.py
@@ -22,6 +22,7 @@
list_files,
list_groups,
list_webhooks,
+ search_files,
set_file_tags,
store_files,
sync,
@@ -64,6 +65,7 @@ def ucare_argparser():
get_file_tags.register_arguments(subparsers)
set_file_tags.register_arguments(subparsers)
update_file_tags.register_arguments(subparsers)
+ search_files.register_arguments(subparsers)
# common arguments
parser.add_argument(
diff --git a/tests/functional/api/cassettes/test_search_files.yaml b/tests/functional/api/cassettes/test_search_files.yaml
new file mode 100644
index 0000000..5c74630
--- /dev/null
+++ b/tests/functional/api/cassettes/test_search_files.yaml
@@ -0,0 +1,31 @@
+interactions:
+- request:
+ body: '{"query": "sunset", "tags": {"all": ["cat"]}, "sort": ["-score"]}'
+ headers:
+ accept:
+ - '*/*'
+ accept-encoding:
+ - gzip, deflate
+ connection:
+ - keep-alive
+ content-type:
+ - application/json
+ host:
+ - api.uploadcare.com
+ method: POST
+ uri: https://api.uploadcare.com/files/search/?limit=2
+ response:
+ content: '{"next":"https://api.uploadcare.com/files/search/?limit=2&offset=2","previous":null,"total":5,"per_page":2,"results":[{"uuid":"a55d6b25-d03c-4038-9838-6e06bb7df598","original_filename":"sunset-cat.jpg","size":3518420,"mime_type":"image/jpeg","is_image":true,"is_ready":true,"datetime_uploaded":"2026-02-16T14:44:29.395043Z","datetime_stored":"2026-02-16T14:44:29.637342Z","datetime_removed":null,"original_file_url":"https://ucarecdn.com/a55d6b25-d03c-4038-9838-6e06bb7df598/sunset-cat.jpg","url":"https://api.uploadcare.com/files/a55d6b25-d03c-4038-9838-6e06bb7df598/","variations":null,"metadata":{"album":"holiday"},"tags":["cat","animal"],"highlight":{"original_filename":["sunset-cat.jpg"],"metadata":{"album":"summer
+ sunset"}}},{"uuid":"1a9c5240-7d9b-4473-851b-45fa4b0bed64","original_filename":"sunset.png","size":1024,"mime_type":"image/png","is_image":true,"is_ready":true,"datetime_uploaded":"2026-02-15T10:00:00Z","datetime_stored":null,"datetime_removed":null,"original_file_url":"https://ucarecdn.com/1a9c5240-7d9b-4473-851b-45fa4b0bed64/sunset.png","url":"https://api.uploadcare.com/files/1a9c5240-7d9b-4473-851b-45fa4b0bed64/","variations":null,"metadata":{},"tags":[],"highlight":{"original_filename":["sunset.png"]}}]}'
+ headers:
+ Connection:
+ - keep-alive
+ Content-Type:
+ - application/vnd.uploadcare-v0.7+json
+ Server:
+ - nginx
+ Vary:
+ - Accept
+ http_version: HTTP/1.1
+ status_code: 200
+version: 1
diff --git a/tests/functional/api/cassettes/test_search_files_empty_result.yaml b/tests/functional/api/cassettes/test_search_files_empty_result.yaml
new file mode 100644
index 0000000..249aeb8
--- /dev/null
+++ b/tests/functional/api/cassettes/test_search_files_empty_result.yaml
@@ -0,0 +1,32 @@
+interactions:
+- request:
+ body: '{"query": "nothing-matches-this"}'
+ headers:
+ accept:
+ - '*/*'
+ accept-encoding:
+ - gzip, deflate
+ connection:
+ - keep-alive
+ content-type:
+ - application/json
+ host:
+ - api.uploadcare.com
+ method: POST
+ uri: https://api.uploadcare.com/files/search/
+ response:
+ content: '{"next":null,"previous":null,"total":0,"per_page":20,"results":[]}'
+ headers:
+ Connection:
+ - keep-alive
+ Content-Length:
+ - '65'
+ Content-Type:
+ - application/vnd.uploadcare-v0.7+json
+ Server:
+ - nginx
+ Vary:
+ - Accept
+ http_version: HTTP/1.1
+ status_code: 200
+version: 1
diff --git a/tests/functional/api/cassettes/test_search_files_with_appdata.yaml b/tests/functional/api/cassettes/test_search_files_with_appdata.yaml
new file mode 100644
index 0000000..f5c0a8c
--- /dev/null
+++ b/tests/functional/api/cassettes/test_search_files_with_appdata.yaml
@@ -0,0 +1,30 @@
+interactions:
+- request:
+ body: '{"tags": {"all": ["cat"]}}'
+ headers:
+ accept:
+ - '*/*'
+ accept-encoding:
+ - gzip, deflate
+ connection:
+ - keep-alive
+ content-type:
+ - application/json
+ host:
+ - api.uploadcare.com
+ method: POST
+ uri: https://api.uploadcare.com/files/search/?include=appdata
+ response:
+ content: '{"next":null,"previous":null,"total":1,"per_page":20,"results":[{"uuid":"a55d6b25-d03c-4038-9838-6e06bb7df598","original_filename":"cat.jpg","size":1024,"mime_type":"image/jpeg","is_image":true,"is_ready":true,"tags":["cat"],"metadata":{},"appdata":{"uc_clamav_virus_scan":{"version":"0.104.2","datetime_created":"2026-02-16T14:44:29.395043Z","datetime_updated":"2026-02-16T14:44:29.637342Z","data":{"infected":false}}}}]}'
+ headers:
+ Connection:
+ - keep-alive
+ Content-Type:
+ - application/vnd.uploadcare-v0.7+json
+ Server:
+ - nginx
+ Vary:
+ - Accept
+ http_version: HTTP/1.1
+ status_code: 200
+version: 1
diff --git a/tests/functional/api/test_search_api.py b/tests/functional/api/test_search_api.py
new file mode 100644
index 0000000..4870077
--- /dev/null
+++ b/tests/functional/api/test_search_api.py
@@ -0,0 +1,111 @@
+"""Response parsing for file search.
+
+Request bodies are asserted in ``test_search_api_requests.py`` instead: VCR
+matches on method and URI only.
+"""
+
+import pytest
+
+from pyuploadcare.api.entities import FileSearchInfo
+from pyuploadcare.api.search_entities import FileSearchRequest, TagsFilter
+
+
+@pytest.mark.vcr
+def test_search_files(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(
+ query="sunset", tags=TagsFilter(all_=["cat"]), sort=["-score"]
+ ),
+ limit=2,
+ )
+
+ assert response.total == 5
+ assert response.per_page == 2
+ assert response.previous is None
+ assert response.next == (
+ "https://api.uploadcare.com/files/search/?limit=2&offset=2"
+ )
+ assert len(response.results) == 2
+
+
+def test_search_files_parses_file_info(uploadcare, vcr):
+ with vcr.use_cassette("test_search_files"):
+ response = uploadcare.search_files(
+ FileSearchRequest(
+ query="sunset", tags=TagsFilter(all_=["cat"]), sort=["-score"]
+ ),
+ limit=2,
+ )
+
+ first = response.results[0]
+
+ assert isinstance(first, FileSearchInfo)
+ assert str(first.uuid) == "a55d6b25-d03c-4038-9838-6e06bb7df598"
+ assert first.original_filename == "sunset-cat.jpg"
+ assert first.size == 3518420
+ assert first.is_image is True
+ assert first.tags == ["cat", "animal"]
+ assert first.metadata == {"album": "holiday"}
+
+
+def test_search_files_parses_highlight(uploadcare, vcr):
+ with vcr.use_cassette("test_search_files"):
+ response = uploadcare.search_files(
+ FileSearchRequest(
+ query="sunset", tags=TagsFilter(all_=["cat"]), sort=["-score"]
+ ),
+ limit=2,
+ )
+
+ highlight = response.results[0].highlight
+
+ assert highlight is not None
+ assert highlight.original_filename == ["sunset-cat.jpg"]
+ # OpenAPI declares `metadata` as an object of plain strings.
+ assert highlight.metadata == {"album": "summer sunset"}
+ # Absent for fields that did not match a full-text condition.
+ assert highlight.detected_mime_type is None
+
+
+def test_search_files_handles_a_result_without_tags_or_highlight(
+ uploadcare, vcr
+):
+ with vcr.use_cassette("test_search_files"):
+ response = uploadcare.search_files(
+ FileSearchRequest(
+ query="sunset", tags=TagsFilter(all_=["cat"]), sort=["-score"]
+ ),
+ limit=2,
+ )
+
+ second = response.results[1]
+
+ assert second.tags == []
+ assert second.datetime_stored is None
+ assert second.highlight is not None
+ assert second.highlight.metadata is None
+
+
+@pytest.mark.vcr
+def test_search_files_empty_result(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(query="nothing-matches-this")
+ )
+
+ assert response.total == 0
+ assert response.results == []
+ assert response.next is None
+
+
+@pytest.mark.vcr
+def test_search_files_with_appdata(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(tags=TagsFilter(all_=["cat"])),
+ include_appdata=True,
+ )
+
+ appdata = response.results[0].appdata
+
+ assert appdata is not None
+ assert appdata.uc_clamav_virus_scan is not None
+ assert appdata.uc_clamav_virus_scan.data.infected is False
diff --git a/tests/functional/api/test_search_api_requests.py b/tests/functional/api/test_search_api_requests.py
new file mode 100644
index 0000000..323f906
--- /dev/null
+++ b/tests/functional/api/test_search_api_requests.py
@@ -0,0 +1,184 @@
+"""Request shape assertions for file search.
+
+VCR matches on method and URI only, so the POST body has to be asserted
+against a mocked client instead of a cassette.
+"""
+
+from typing import Any, Dict
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pydantic import ValidationError
+
+from pyuploadcare.api.api import SEARCH_MAX_LIMIT, SEARCH_MAX_WINDOW
+from pyuploadcare.api.search_entities import (
+ FileSearchRequest,
+ SearchExact,
+ TagsFilter,
+)
+from pyuploadcare.exceptions import InvalidParamError
+
+
+SEARCH_URL = "https://api.uploadcare.com/files/search/"
+
+EMPTY_PAGE: Dict[str, Any] = {
+ "next": None,
+ "previous": None,
+ "total": 0,
+ "per_page": 20,
+ "results": [],
+}
+
+
+def _json_response(payload=None):
+ response = MagicMock()
+ response.json.return_value = payload or EMPTY_PAGE
+ return response
+
+
+@pytest.fixture
+def files_api(uploadcare):
+ return uploadcare.files_api
+
+
+def test_search_posts_to_the_search_url(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(FileSearchRequest(query="sunset"))
+
+ mocked_post.assert_called_once_with(SEARCH_URL, json={"query": "sunset"})
+
+
+def test_search_sends_limit_and_offset_as_query_parameters(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(
+ FileSearchRequest(query="sunset"), limit=50, offset=100
+ )
+
+ url = mocked_post.call_args.args[0]
+ assert url == f"{SEARCH_URL}?limit=50&offset=100"
+
+
+def test_search_sends_include_appdata(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(
+ FileSearchRequest(query="sunset"), include_appdata=True
+ )
+
+ assert mocked_post.call_args.args[0] == f"{SEARCH_URL}?include=appdata"
+
+
+def test_search_omits_pagination_parameters_when_not_given(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(FileSearchRequest(query="sunset"))
+
+ assert mocked_post.call_args.args[0] == SEARCH_URL
+
+
+def test_search_accepts_a_dict_request(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search({"tags": {"all": ["cat", "Cat"]}})
+
+ assert mocked_post.call_args.kwargs["json"] == {"tags": {"all": ["cat"]}}
+
+
+def test_search_sends_bracketed_metadata_keys(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(
+ FileSearchRequest(exact=SearchExact(metadata={"color": ["red"]}))
+ )
+
+ assert mocked_post.call_args.kwargs["json"] == {
+ "exact": {"metadata[color]": ["red"]}
+ }
+
+
+def test_search_sends_tag_filter_wire_names(files_api):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(
+ FileSearchRequest(tags=TagsFilter(any_=["cat"], none_=["dog"]))
+ )
+
+ assert mocked_post.call_args.kwargs["json"] == {
+ "tags": {"any": ["cat"], "none": ["dog"]}
+ }
+
+
+def test_search_rejects_an_invalid_dict_before_any_request(files_api):
+ with patch.object(files_api._client, "post") as mocked_post:
+ with pytest.raises(ValidationError):
+ files_api.search({})
+
+ mocked_post.assert_not_called()
+
+
+@pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"limit": 0},
+ {"limit": -1},
+ {"limit": SEARCH_MAX_LIMIT + 1},
+ {"offset": -1},
+ {"offset": SEARCH_MAX_WINDOW, "limit": 1},
+ {"offset": SEARCH_MAX_WINDOW - 10},
+ {"offset": SEARCH_MAX_WINDOW - 10, "limit": 20},
+ ],
+)
+def test_search_rejects_out_of_range_pagination(files_api, kwargs):
+ with patch.object(files_api._client, "post") as mocked_post:
+ with pytest.raises(InvalidParamError):
+ files_api.search(FileSearchRequest(query="sunset"), **kwargs)
+
+ mocked_post.assert_not_called()
+
+
+@pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"limit": 1.5},
+ {"limit": True},
+ {"limit": "10"},
+ {"offset": 1.5},
+ {"offset": False},
+ {"offset": "0"},
+ ],
+)
+def test_search_rejects_non_integer_pagination(files_api, kwargs):
+ """Annotations are not enforced at runtime and `bool` is an `int`."""
+ with patch.object(files_api._client, "post") as mocked_post:
+ with pytest.raises(InvalidParamError):
+ files_api.search(FileSearchRequest(query="sunset"), **kwargs)
+
+ mocked_post.assert_not_called()
+
+
+@pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"limit": 1},
+ {"limit": SEARCH_MAX_LIMIT},
+ {"offset": 0},
+ {"offset": SEARCH_MAX_WINDOW - 1, "limit": 1},
+ {"offset": SEARCH_MAX_WINDOW - 20},
+ ],
+)
+def test_search_accepts_pagination_at_the_boundaries(files_api, kwargs):
+ with patch.object(
+ files_api._client, "post", return_value=_json_response()
+ ) as mocked_post:
+ files_api.search(FileSearchRequest(query="sunset"), **kwargs)
+
+ mocked_post.assert_called_once()
diff --git a/tests/functional/api/test_search_request.py b/tests/functional/api/test_search_request.py
new file mode 100644
index 0000000..5586276
--- /dev/null
+++ b/tests/functional/api/test_search_request.py
@@ -0,0 +1,361 @@
+"""Validation and serialization of file search request models."""
+
+from datetime import datetime, timezone
+
+import pytest
+from pydantic import ValidationError
+
+from pyuploadcare.api.search_entities import (
+ MAX_SORT_KEYS,
+ DatetimeRange,
+ FileSearchRequest,
+ SearchExact,
+ SearchPhrase,
+ SearchSort,
+ SizeRange,
+ TagsFilter,
+)
+from pyuploadcare.exceptions import MetadataValidationError, TagValidationError
+
+
+# --- at least one condition -------------------------------------------------
+
+
+def test_request_requires_at_least_one_condition():
+ with pytest.raises(ValidationError):
+ FileSearchRequest()
+
+
+def test_modifiers_alone_are_not_a_condition():
+ """`fuzziness` and `sort` are modifiers, not conditions."""
+ with pytest.raises(ValidationError):
+ FileSearchRequest(fuzziness=True, sort=[SearchSort.SCORE])
+
+
+@pytest.mark.parametrize(
+ "condition",
+ [
+ {"query": "sunset"},
+ {"phrase": SearchPhrase(original_filename="sunset")},
+ {"exact": SearchExact(original_filename=["sunset.jpg"])},
+ {"datetime_uploaded": DatetimeRange(gt=datetime(2024, 1, 1))},
+ {"size": SizeRange(gt=1000)},
+ {"is_image": True},
+ {"is_image": False},
+ {"tags": TagsFilter(any_=["cat"])},
+ ],
+)
+def test_any_single_condition_is_enough(condition):
+ FileSearchRequest(**condition)
+
+
+# --- query and phrase -------------------------------------------------------
+
+
+def test_query_requires_four_characters():
+ with pytest.raises(ValidationError):
+ FileSearchRequest(query="abc")
+
+
+def test_query_accepts_four_characters():
+ assert FileSearchRequest(query="abcd").query == "abcd"
+
+
+def test_phrase_requires_at_least_one_field():
+ with pytest.raises(ValidationError):
+ SearchPhrase()
+
+
+@pytest.mark.parametrize(
+ "field", ["original_filename", "metadata", "detected_mime_type"]
+)
+def test_phrase_values_require_four_characters(field):
+ with pytest.raises(ValidationError):
+ SearchPhrase(**{field: "abc"})
+
+
+# --- exact ------------------------------------------------------------------
+
+
+def test_exact_requires_at_least_one_condition():
+ with pytest.raises(ValidationError):
+ SearchExact()
+
+
+def test_exact_rejects_empty_metadata_mapping():
+ """It would otherwise serialize to an empty `"exact": {}`."""
+ with pytest.raises(ValidationError):
+ SearchExact(metadata={})
+
+
+@pytest.mark.parametrize(
+ "field", ["uuid", "original_filename", "detected_mime_type"]
+)
+def test_exact_rejects_empty_arrays(field):
+ with pytest.raises(ValidationError):
+ SearchExact(**{field: []})
+
+
+def test_exact_rejects_empty_metadata_value_array():
+ with pytest.raises(ValidationError):
+ SearchExact(metadata={"color": []})
+
+
+def test_exact_rejects_invalid_metadata_key():
+ """A `[` or `]` in a key could forge a different wire key."""
+ with pytest.raises(MetadataValidationError):
+ SearchExact(metadata={"bad]key[": ["red"]})
+
+
+# --- phrase / exact overlap -------------------------------------------------
+
+
+@pytest.mark.parametrize("field", ["original_filename", "detected_mime_type"])
+def test_field_cannot_be_in_both_phrase_and_exact(field):
+ with pytest.raises(ValidationError):
+ FileSearchRequest(
+ phrase=SearchPhrase(**{field: "sunset"}),
+ exact=SearchExact(**{field: ["sunset"]}),
+ )
+
+
+def test_different_fields_in_phrase_and_exact_are_allowed():
+ FileSearchRequest(
+ phrase=SearchPhrase(original_filename="sunset"),
+ exact=SearchExact(uuid=["a55d6b25-d03c-4038-9838-6e06bb7df598"]),
+ )
+
+
+def test_metadata_in_both_phrase_and_exact_is_allowed():
+ """`phrase.metadata` and `exact.metadata[key]` are distinct wire keys."""
+ FileSearchRequest(
+ phrase=SearchPhrase(metadata="sunset"),
+ exact=SearchExact(metadata={"color": ["red"]}),
+ )
+
+
+# --- sort -------------------------------------------------------------------
+
+
+def test_sort_rejects_empty_list():
+ with pytest.raises(ValidationError):
+ FileSearchRequest(query="sunset", sort=[])
+
+
+def test_sort_rejects_too_many_keys():
+ keys = [
+ SearchSort.SCORE,
+ SearchSort.SIZE,
+ SearchSort.DATETIME_UPLOADED,
+ SearchSort.ORIGINAL_FILENAME,
+ SearchSort.SCORE_DESC,
+ ]
+ assert len(keys) == MAX_SORT_KEYS + 1
+
+ with pytest.raises(ValidationError):
+ FileSearchRequest(query="sunset", sort=keys)
+
+
+def test_sort_rejects_duplicate_keys():
+ with pytest.raises(ValidationError):
+ FileSearchRequest(
+ query="sunset", sort=[SearchSort.SIZE, SearchSort.SIZE]
+ )
+
+
+def test_sort_rejects_both_directions_of_the_same_key():
+ with pytest.raises(ValidationError):
+ FileSearchRequest(
+ query="sunset", sort=[SearchSort.SCORE, SearchSort.SCORE_DESC]
+ )
+
+
+def test_sort_rejects_unknown_key():
+ with pytest.raises(ValidationError):
+ FileSearchRequest(query="sunset", sort=["relevance"])
+
+
+def test_sort_accepts_plain_strings():
+ request = FileSearchRequest(query="sunset", sort=["-score", "size"])
+ assert request.sort == [SearchSort.SCORE_DESC, SearchSort.SIZE]
+
+
+# --- tags -------------------------------------------------------------------
+
+
+def test_tags_filter_requires_at_least_one_list():
+ with pytest.raises(ValidationError):
+ TagsFilter()
+
+
+def test_tags_filter_rejects_only_empty_lists():
+ with pytest.raises(ValidationError):
+ TagsFilter(any_=[], all_=[])
+
+
+def test_tags_filter_accepts_wire_names():
+ tags = TagsFilter(**{"any": ["cat"], "none": ["dog"]})
+ assert tags.any_ == ["cat"]
+ assert tags.none_ == ["dog"]
+
+
+def test_tags_filter_normalizes_tags():
+ assert TagsFilter(any_=[" Cat ", "CAT", "dog"]).any_ == ["cat", "dog"]
+
+
+def test_tags_filter_rejects_invalid_tags():
+ with pytest.raises(TagValidationError):
+ TagsFilter(any_=["not valid"])
+
+
+def test_tags_filter_has_no_count_limit():
+ """The 50-tag limit is per stored file, not per filter."""
+ tags = [f"tag{index}" for index in range(60)]
+ assert TagsFilter(any_=tags).any_ == tags
+
+
+# --- ranges -----------------------------------------------------------------
+
+
+@pytest.mark.parametrize("model", [DatetimeRange, SizeRange])
+def test_range_requires_at_least_one_bound(model):
+ with pytest.raises(ValidationError):
+ model()
+
+
+def test_size_range_rejects_negative_values():
+ with pytest.raises(ValidationError):
+ SizeRange(gt=-1)
+
+
+# --- strict scalars ---------------------------------------------------------
+
+
+@pytest.mark.parametrize("value", [True, False, "1000", 1000.5])
+def test_size_range_rejects_non_integers(value):
+ """`gt=True` must not silently become `gt=1`."""
+ with pytest.raises(ValidationError):
+ SizeRange(gt=value)
+
+
+def test_size_range_accepts_integers():
+ size = SizeRange(gt=0, lte=1000)
+ assert (size.gt, size.lte) == (0, 1000)
+
+
+@pytest.mark.parametrize("field", ["is_image", "fuzziness"])
+@pytest.mark.parametrize("value", [1, 0, "true", "false", "yes"])
+def test_boolean_fields_reject_non_booleans(field, value):
+ with pytest.raises(ValidationError):
+ FileSearchRequest(query="sunset", **{field: value})
+
+
+@pytest.mark.parametrize("field", ["is_image", "fuzziness"])
+@pytest.mark.parametrize("value", [True, False])
+def test_boolean_fields_accept_booleans(field, value):
+ request = FileSearchRequest(query="sunset", **{field: value})
+ assert getattr(request, field) is value
+
+
+# --- extra fields -----------------------------------------------------------
+
+
+def test_unknown_field_is_rejected():
+ """A typo must be reported, not silently dropped."""
+ with pytest.raises(ValidationError):
+ FileSearchRequest.model_validate({"querry": "sunset"})
+
+
+def test_unknown_nested_field_is_rejected():
+ with pytest.raises(ValidationError):
+ FileSearchRequest.model_validate(
+ {"exact": {"filename": ["sunset.jpg"]}}
+ )
+
+
+def test_wire_shape_for_exact_metadata_is_rejected():
+ """Dicts use the SDK shape, not the `metadata[key]` wire shape."""
+ with pytest.raises(ValidationError):
+ FileSearchRequest.model_validate(
+ {"exact": {"metadata[color]": ["red"]}}
+ )
+
+
+# --- to_payload -------------------------------------------------------------
+
+
+def test_payload_lifts_exact_metadata_into_bracketed_keys():
+ request = FileSearchRequest(
+ exact=SearchExact(metadata={"color": ["red"], "size": ["xl"]})
+ )
+
+ assert request.to_payload() == {
+ "exact": {"metadata[color]": ["red"], "metadata[size]": ["xl"]}
+ }
+
+
+def test_payload_keeps_other_exact_fields_alongside_metadata():
+ request = FileSearchRequest(
+ exact=SearchExact(
+ original_filename=["sunset.jpg"], metadata={"color": ["red"]}
+ )
+ )
+
+ assert request.to_payload() == {
+ "exact": {
+ "original_filename": ["sunset.jpg"],
+ "metadata[color]": ["red"],
+ }
+ }
+
+
+def test_payload_renders_datetime_as_iso8601():
+ request = FileSearchRequest(
+ datetime_uploaded=DatetimeRange(
+ gte=datetime(2024, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
+ )
+ )
+
+ payload = request.to_payload()
+ assert payload["datetime_uploaded"]["gte"].startswith(
+ "2024-01-02T03:04:05"
+ )
+
+
+def test_payload_renders_sort_keys_as_values():
+ request = FileSearchRequest(
+ query="sunset", sort=[SearchSort.SCORE_DESC, SearchSort.SIZE]
+ )
+
+ assert request.to_payload()["sort"] == ["-score", "size"]
+
+
+def test_payload_uses_tag_filter_wire_names():
+ request = FileSearchRequest(
+ tags=TagsFilter(any_=["cat"], all_=["pet"], none_=["dog"])
+ )
+
+ assert request.to_payload()["tags"] == {
+ "any": ["cat"],
+ "all": ["pet"],
+ "none": ["dog"],
+ }
+
+
+def test_payload_drops_empty_tag_lists():
+ request = FileSearchRequest(tags=TagsFilter(any_=["cat"], none_=[]))
+
+ assert request.to_payload()["tags"] == {"any": ["cat"]}
+
+
+def test_payload_omits_unset_fields():
+ request = FileSearchRequest(query="sunset")
+
+ assert request.to_payload() == {"query": "sunset"}
+
+
+def test_payload_keeps_false_values():
+ """`exclude_none` must not drop `is_image=False` or `fuzziness=False`."""
+ request = FileSearchRequest(is_image=False, fuzziness=False)
+
+ assert request.to_payload() == {"is_image": False, "fuzziness": False}
diff --git a/tests/functional/resources/test_search_pagination.py b/tests/functional/resources/test_search_pagination.py
new file mode 100644
index 0000000..6acb673
--- /dev/null
+++ b/tests/functional/resources/test_search_pagination.py
@@ -0,0 +1,336 @@
+"""Pagination behaviour of ``Uploadcare.iterate_search_files``."""
+
+from unittest.mock import patch
+from uuid import UUID
+
+import pytest
+from pydantic import ValidationError
+
+from pyuploadcare.api.api import (
+ SEARCH_DEFAULT_LIMIT,
+ SEARCH_MAX_LIMIT,
+ SEARCH_MAX_WINDOW,
+)
+from pyuploadcare.api.responses import FileSearchResponse
+from pyuploadcare.exceptions import InvalidParamError
+
+
+# Filter-only requests have an undefined order, so every request used for
+# paging carries an explicit `sort`. See `test_undefined_order_warning`.
+REQUEST = {"tags": {"all": ["cat"]}, "sort": ["-datetime_uploaded"]}
+
+# A `next` URL pointing somewhere else entirely. The iterator must never
+# request it: the REST client attaches credentials to any URL it is given.
+HOSTILE_NEXT = "https://evil.example/files/search/?limit=2&offset=999"
+
+
+def _uuid(index: int) -> str:
+ return str(UUID(int=index))
+
+
+def _page(count, next_url=None, total=None, offset=0, per_page=None):
+ return FileSearchResponse.model_validate(
+ {
+ "next": next_url,
+ "previous": None,
+ "total": count if total is None else total,
+ "per_page": count if per_page is None else per_page,
+ "results": [
+ {"uuid": _uuid(offset + index)} for index in range(count)
+ ],
+ }
+ )
+
+
+def _patched_search(uploadcare, pages):
+ return patch.object(uploadcare.files_api, "search", side_effect=pages)
+
+
+def _offsets(mocked_search):
+ return [call.kwargs["offset"] for call in mocked_search.call_args_list]
+
+
+def _limits(mocked_search):
+ return [call.kwargs["limit"] for call in mocked_search.call_args_list]
+
+
+def test_single_page_is_yielded(uploadcare):
+ with _patched_search(uploadcare, [_page(3)]):
+ results = list(uploadcare.iterate_search_files(REQUEST))
+
+ assert len(results) == 3
+
+
+def test_follows_pages_until_next_is_none(uploadcare):
+ pages = [
+ _page(2, next_url=HOSTILE_NEXT, total=4),
+ _page(2, next_url=None, total=4, offset=2),
+ ]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ results = list(
+ uploadcare.iterate_search_files(REQUEST, request_limit=2)
+ )
+
+ assert [str(item.uuid) for item in results] == [_uuid(i) for i in range(4)]
+ assert _offsets(mocked_search) == [0, 2]
+
+
+def test_offset_advances_by_the_requested_page_size(uploadcare):
+ """Offset pagination advances by the window, not by results received.
+
+ A short page that still reports a `next` would otherwise make the
+ following request overlap it and repeat files.
+ """
+ pages = [
+ # Asked for 20, got 3, but there is more.
+ _page(3, next_url=HOSTILE_NEXT, total=99, per_page=20),
+ _page(0, next_url=None, total=99, per_page=20),
+ ]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ list(uploadcare.iterate_search_files(REQUEST))
+
+ assert _offsets(mocked_search) == [0, SEARCH_DEFAULT_LIMIT]
+
+
+def test_next_url_is_never_requested(uploadcare):
+ """Offsets are computed locally, never taken from the `next` URL."""
+ pages = [
+ _page(2, next_url=HOSTILE_NEXT, total=4),
+ _page(2, next_url=None, total=4, offset=2),
+ ]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ with patch.object(uploadcare.rest_client, "post") as mocked_post:
+ list(uploadcare.iterate_search_files(REQUEST, request_limit=2))
+
+ mocked_post.assert_not_called()
+ # `HOSTILE_NEXT` advertises offset 999 on another origin; ignored.
+ assert _offsets(mocked_search) == [0, 2]
+
+
+def test_stops_on_an_empty_page(uploadcare):
+ pages = [
+ _page(2, next_url=HOSTILE_NEXT, total=99),
+ _page(0, next_url=HOSTILE_NEXT, total=99),
+ ]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ results = list(
+ uploadcare.iterate_search_files(REQUEST, request_limit=2)
+ )
+
+ assert len(results) == 2
+ assert mocked_search.call_count == 2
+
+
+def test_does_not_stop_on_an_understated_total(uploadcare):
+ """`total` is documented as possibly approximate, so it is not a stop."""
+ pages = [
+ _page(2, next_url=HOSTILE_NEXT, total=2),
+ _page(2, next_url=None, total=2, offset=2),
+ ]
+
+ with _patched_search(uploadcare, pages):
+ results = list(
+ uploadcare.iterate_search_files(REQUEST, request_limit=2)
+ )
+
+ assert len(results) == 4
+
+
+def test_limit_caps_the_total_yielded(uploadcare):
+ pages = [_page(2, next_url=HOSTILE_NEXT, total=99)]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ results = list(uploadcare.iterate_search_files(REQUEST, limit=2))
+
+ assert len(results) == 2
+ assert mocked_search.call_count == 1
+
+
+def test_limit_smaller_than_a_page_truncates_mid_page(uploadcare):
+ pages = [_page(5, next_url=HOSTILE_NEXT, total=99)]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ results = list(uploadcare.iterate_search_files(REQUEST, limit=3))
+
+ assert len(results) == 3
+ # The page size is clamped to the remaining amount.
+ assert _limits(mocked_search) == [3]
+
+
+def test_limit_zero_makes_no_request(uploadcare):
+ with patch.object(uploadcare.files_api, "search") as mocked_search:
+ results = list(uploadcare.iterate_search_files(REQUEST, limit=0))
+
+ assert results == []
+ mocked_search.assert_not_called()
+
+
+def test_default_page_size(uploadcare):
+ with _patched_search(uploadcare, [_page(1)]) as mocked_search:
+ list(uploadcare.iterate_search_files(REQUEST))
+
+ assert _limits(mocked_search) == [SEARCH_DEFAULT_LIMIT]
+
+
+def test_request_limit_sets_the_page_size(uploadcare):
+ with _patched_search(uploadcare, [_page(1)]) as mocked_search:
+ list(uploadcare.iterate_search_files(REQUEST, request_limit=7))
+
+ assert _limits(mocked_search) == [7]
+
+
+def test_offset_is_forwarded(uploadcare):
+ with _patched_search(uploadcare, [_page(1)]) as mocked_search:
+ list(uploadcare.iterate_search_files(REQUEST, offset=42))
+
+ assert _offsets(mocked_search) == [42]
+
+
+@pytest.mark.parametrize(
+ "offset, expected_page_size",
+ [
+ (SEARCH_MAX_WINDOW - 10, 10),
+ (SEARCH_MAX_WINDOW - 50, 20),
+ (SEARCH_MAX_WINDOW - 1, 1),
+ ],
+)
+def test_page_size_is_clamped_to_the_search_window(
+ uploadcare, offset, expected_page_size
+):
+ """`offset` + `limit` must never exceed the window."""
+ with _patched_search(uploadcare, [_page(1, offset=offset)]) as mocked:
+ list(uploadcare.iterate_search_files(REQUEST, offset=offset))
+
+ call = mocked.call_args_list[0]
+ assert call.kwargs["limit"] == expected_page_size
+ assert call.kwargs["offset"] + call.kwargs["limit"] <= SEARCH_MAX_WINDOW
+
+
+def test_stops_at_the_search_window(uploadcare):
+ """Paging cannot go past the first 1000 results."""
+ offset = SEARCH_MAX_WINDOW - 2
+ pages = [_page(2, next_url=HOSTILE_NEXT, total=99, offset=offset)]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ results = list(uploadcare.iterate_search_files(REQUEST, offset=offset))
+
+ assert len(results) == 2
+ assert mocked_search.call_count == 1
+
+
+def test_offset_at_the_window_makes_no_request(uploadcare):
+ with patch.object(uploadcare.files_api, "search") as mocked_search:
+ results = list(
+ uploadcare.iterate_search_files(REQUEST, offset=SEARCH_MAX_WINDOW)
+ )
+
+ assert results == []
+ mocked_search.assert_not_called()
+
+
+# --- argument and request validation ----------------------------------------
+
+
+@pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"limit": -1},
+ {"request_limit": 0},
+ {"request_limit": -1},
+ {"request_limit": SEARCH_MAX_LIMIT + 1},
+ {"offset": -1},
+ {"offset": SEARCH_MAX_WINDOW + 1},
+ ],
+)
+def test_rejects_out_of_range_arguments(uploadcare, kwargs):
+ with patch.object(uploadcare.files_api, "search") as mocked_search:
+ with pytest.raises(InvalidParamError):
+ uploadcare.iterate_search_files(REQUEST, **kwargs)
+
+ mocked_search.assert_not_called()
+
+
+@pytest.mark.parametrize(
+ "kwargs",
+ [
+ {"limit": 1.5},
+ {"limit": True},
+ {"request_limit": "10"},
+ {"request_limit": False},
+ {"offset": 2.5},
+ ],
+)
+def test_rejects_non_integer_arguments(uploadcare, kwargs):
+ with patch.object(uploadcare.files_api, "search") as mocked_search:
+ with pytest.raises(InvalidParamError):
+ uploadcare.iterate_search_files(REQUEST, **kwargs)
+
+ mocked_search.assert_not_called()
+
+
+def test_argument_errors_are_raised_eagerly(uploadcare):
+ """Not deferred until the returned iterator is first advanced."""
+ with pytest.raises(InvalidParamError):
+ uploadcare.iterate_search_files(REQUEST, limit=-1)
+
+
+def test_invalid_request_is_rejected_eagerly(uploadcare):
+ """The request itself is validated by the wrapper, not by the generator."""
+ with patch.object(uploadcare.files_api, "search") as mocked_search:
+ with pytest.raises(ValidationError):
+ uploadcare.iterate_search_files({})
+
+ mocked_search.assert_not_called()
+
+
+def test_invalid_request_is_rejected_even_with_limit_zero(uploadcare):
+ """`limit=0` short-circuits paging but must not skip validation."""
+ with pytest.raises(ValidationError):
+ uploadcare.iterate_search_files({}, limit=0)
+
+
+def test_request_is_validated_only_once(uploadcare):
+ """The generator receives a model, so pages do not re-validate."""
+ pages = [
+ _page(2, next_url=HOSTILE_NEXT, total=4),
+ _page(2, next_url=None, total=4, offset=2),
+ ]
+
+ with _patched_search(uploadcare, pages) as mocked_search:
+ list(uploadcare.iterate_search_files(REQUEST, request_limit=2))
+
+ requests = [call.args[0] for call in mocked_search.call_args_list]
+ assert len(requests) == 2
+ assert requests[0] is requests[1]
+
+
+# --- undefined result order -------------------------------------------------
+
+
+def test_undefined_order_warning(uploadcare):
+ """A filter-only request without `sort` has an undefined order."""
+ with _patched_search(uploadcare, [_page(1)]):
+ with pytest.warns(UserWarning, match="undefined"):
+ list(uploadcare.iterate_search_files({"tags": {"all": ["cat"]}}))
+
+
+@pytest.mark.parametrize(
+ "request_",
+ [
+ {"tags": {"all": ["cat"]}, "sort": ["-datetime_uploaded"]},
+ {"query": "sunset"},
+ {"phrase": {"original_filename": "sunset"}},
+ ],
+)
+def test_no_warning_when_the_order_is_defined(uploadcare, request_):
+ """`sort`, or relevance from `query`/`phrase`, gives a defined order."""
+ import warnings
+
+ with _patched_search(uploadcare, [_page(1)]):
+ with warnings.catch_warnings():
+ warnings.simplefilter("error")
+ list(uploadcare.iterate_search_files(request_))
diff --git a/tests/functional/ucare_cli/cassettes/test_cli_search_files.yaml b/tests/functional/ucare_cli/cassettes/test_cli_search_files.yaml
new file mode 100644
index 0000000..4228def
--- /dev/null
+++ b/tests/functional/ucare_cli/cassettes/test_cli_search_files.yaml
@@ -0,0 +1,30 @@
+interactions:
+- request:
+ body: '{"query": "sunset", "tags": {"any": ["cat"]}}'
+ headers:
+ accept:
+ - '*/*'
+ accept-encoding:
+ - gzip, deflate
+ connection:
+ - keep-alive
+ content-type:
+ - application/json
+ host:
+ - api.uploadcare.com
+ method: POST
+ uri: https://api.uploadcare.com/files/search/?limit=1
+ response:
+ content: '{"next":null,"previous":null,"total":1,"per_page":1,"results":[{"uuid":"a55d6b25-d03c-4038-9838-6e06bb7df598","original_filename":"sunset-cat.jpg","size":1024,"mime_type":"image/jpeg","is_image":true,"is_ready":true,"tags":["cat"],"metadata":{},"highlight":{"original_filename":["sunset-cat.jpg"]}}]}'
+ headers:
+ Connection:
+ - keep-alive
+ Content-Type:
+ - application/vnd.uploadcare-v0.7+json
+ Server:
+ - nginx
+ Vary:
+ - Accept
+ http_version: HTTP/1.1
+ status_code: 200
+version: 1
diff --git a/tests/functional/ucare_cli/test_search_files.py b/tests/functional/ucare_cli/test_search_files.py
new file mode 100644
index 0000000..af5c83d
--- /dev/null
+++ b/tests/functional/ucare_cli/test_search_files.py
@@ -0,0 +1,104 @@
+from datetime import datetime
+
+import pytest
+from tests.functional.ucare_cli.helpers import arg_namespace
+
+from pyuploadcare.exceptions import InvalidParamError
+from pyuploadcare.ucare_cli.commands.search_files import (
+ _build_request,
+ search_files,
+)
+from pyuploadcare.ucare_cli.main import main
+
+
+@pytest.mark.vcr
+def test_cli_search_files(capsys, uploadcare):
+ search_files(
+ arg_namespace("search_files --query sunset --tags_any cat --limit 1"),
+ uploadcare,
+ )
+ captured = capsys.readouterr()
+
+ assert '"total": 1' in captured.out
+ assert '"sunset-cat.jpg"' in captured.out
+ assert '"sunset-cat.jpg"' in captured.out
+
+
+def test_cli_search_files_without_conditions_prints_error(capsys):
+ """A `ValidationError` must not surface as a traceback."""
+ main(
+ arg_namespace(
+ "--pub_key demopublickey --secret demosecretkey search_files"
+ ),
+ config_file_names=(),
+ )
+ captured = capsys.readouterr()
+
+ assert "ERROR:" in captured.out
+
+
+def test_cli_search_files_without_conditions_raises(uploadcare):
+ with pytest.raises(InvalidParamError):
+ search_files(arg_namespace("search_files"), uploadcare)
+
+
+def test_cli_builds_nested_conditions():
+ parsed = arg_namespace(
+ "search_files"
+ " --phrase_original_filename sunset"
+ " --exact_detected_mime_type image/jpeg image/png"
+ " --size_gt 1000 --size_lte 5000"
+ " --uploaded_gte 2024-01-02"
+ " --tags_any cat dog --tags_none old"
+ " --is_image true"
+ " --fuzziness"
+ )
+
+ assert _build_request(parsed) == {
+ "fuzziness": True,
+ "is_image": True,
+ "phrase": {"original_filename": "sunset"},
+ "exact": {"detected_mime_type": ["image/jpeg", "image/png"]},
+ "size": {"gt": 1000, "lte": 5000},
+ "datetime_uploaded": {"gte": datetime(2024, 1, 2)},
+ "tags": {"any": ["cat", "dog"], "none": ["old"]},
+ }
+
+
+def test_cli_omits_unset_conditions():
+ parsed = arg_namespace("search_files --query sunset")
+
+ assert _build_request(parsed) == {"query": "sunset"}
+
+
+def test_cli_sort_accepts_repeated_keys():
+ parsed = arg_namespace("search_files --query sunset --sort=-score")
+ parsed_two = arg_namespace(
+ "search_files --query sunset --sort=-score --sort=size"
+ )
+
+ assert parsed.sort == ["-score"]
+ assert parsed_two.sort == ["-score", "size"]
+
+
+def test_cli_sort_rejects_unknown_key():
+ with pytest.raises(SystemExit):
+ arg_namespace("search_files --query sunset --sort=relevance")
+
+
+@pytest.mark.parametrize("value", ["true", "false", "TRUE", " False "])
+def test_cli_is_image_accepts_booleans(value):
+ parsed = arg_namespace(["search_files", "--is_image", value])
+ assert parsed.is_image is (value.strip().lower() == "true")
+
+
+@pytest.mark.parametrize("value", ["yes", "1", "maybe", ""])
+def test_cli_is_image_rejects_other_values(value):
+ """Unlike `bool_or_none`, unknown input must not become `None`."""
+ with pytest.raises(SystemExit):
+ arg_namespace(["search_files", "--is_image", value])
+
+
+def test_cli_uploaded_bound_rejects_unparsable_datetime():
+ with pytest.raises(SystemExit):
+ arg_namespace("search_files --uploaded_gte not-a-date")
diff --git a/tests/integration/test_file_search.py b/tests/integration/test_file_search.py
new file mode 100644
index 0000000..d6f3835
--- /dev/null
+++ b/tests/integration/test_file_search.py
@@ -0,0 +1,247 @@
+"""File search against the live REST API.
+
+https://uploadcare.com/docs/file-search/
+
+Search indexing is asynchronous, so tests that need to find a file they just
+uploaded poll through ``wait_until_searchable``. Everything that only needs
+*some* indexed content searches over what the project already holds.
+"""
+
+import pytest
+
+from pyuploadcare.api.entities import FileSearchInfo
+from pyuploadcare.api.search_entities import (
+ FileSearchRequest,
+ SearchExact,
+ SearchPhrase,
+ SizeRange,
+ TagsFilter,
+)
+
+from .utils import (
+ search_term,
+ unique_tag,
+ upload_image_file,
+ wait_until_searchable,
+)
+
+
+# Filter-only requests have an undefined order, so anything paginated or
+# order-sensitive carries an explicit sort.
+BY_NEWEST = ["-datetime_uploaded"]
+
+
+@pytest.fixture(scope="module")
+def indexed_file(uploadcare):
+ """A file already in the project, and therefore already searchable."""
+ files = list(uploadcare.list_files(limit=1, removed=False))
+
+ if not files:
+ pytest.skip("the project has no files to search")
+
+ return files[0]
+
+
+@pytest.fixture(scope="module")
+def indexed_term(indexed_file):
+ term = search_term(indexed_file.info.get("original_filename"))
+
+ if not term:
+ pytest.skip("no filename with a 4+ character term to search for")
+
+ return term
+
+
+def test_search_response_shape(uploadcare):
+ """The envelope and every result parse into the SDK's models."""
+ response = uploadcare.search_files(
+ FileSearchRequest(is_image=True, sort=BY_NEWEST), limit=3
+ )
+
+ assert isinstance(response.total, int)
+ assert response.per_page == 3
+ assert len(response.results) <= 3
+
+ for result in response.results:
+ assert isinstance(result, FileSearchInfo)
+ assert isinstance(result.tags, list)
+ assert result.uuid is not None
+
+
+def test_query_returns_a_highlight(uploadcare, indexed_term):
+ response = uploadcare.search_files(
+ FileSearchRequest(query=indexed_term), limit=3
+ )
+
+ assert response.total >= 1
+
+ highlighted = [
+ value
+ for result in response.results
+ if result.highlight and result.highlight.original_filename
+ for value in result.highlight.original_filename
+ ]
+
+ assert highlighted, "expected a highlight for a full text match"
+ assert any("" in value for value in highlighted)
+
+
+def test_phrase_on_original_filename(uploadcare, indexed_term):
+ response = uploadcare.search_files(
+ FileSearchRequest(phrase=SearchPhrase(original_filename=indexed_term)),
+ limit=3,
+ )
+
+ assert response.total >= 1
+
+
+def test_filter_only_search_highlights_nothing(uploadcare):
+ """No highlight field is populated without a full text condition.
+
+ The API reference says `highlight` is "absent for filter-only matches",
+ but the live API sends an empty object instead, so the model parses it
+ into a `SearchHighlight` whose every field is `None`. Either shape is
+ handled; what matters is that no field carries a value.
+ """
+ response = uploadcare.search_files(
+ FileSearchRequest(is_image=True, sort=BY_NEWEST), limit=3
+ )
+
+ if not response.results:
+ pytest.skip("the project has no images to search")
+
+ for result in response.results:
+ if result.highlight is None:
+ continue
+
+ assert result.highlight.original_filename is None
+ assert result.highlight.detected_mime_type is None
+ assert result.highlight.metadata is None
+
+
+def test_exact_uuid(uploadcare, indexed_file):
+ response = uploadcare.search_files(
+ FileSearchRequest(exact=SearchExact(uuid=[indexed_file.uuid]))
+ )
+
+ assert response.total == 1
+ assert str(response.results[0].uuid) == indexed_file.uuid
+
+
+def test_tags_filter_finds_a_tagged_file(uploadcare):
+ """Tags attached on upload become searchable once indexed."""
+ tag = unique_tag()
+ file_ = upload_image_file(uploadcare, tags=[tag])
+
+ try:
+ response = wait_until_searchable(
+ uploadcare,
+ FileSearchRequest(tags=TagsFilter(all_=[tag]), sort=BY_NEWEST),
+ )
+
+ assert response.total == 1
+
+ result = response.results[0]
+ assert str(result.uuid) == file_.uuid
+ assert result.tags == [tag]
+ finally:
+ file_.delete()
+
+
+def test_tags_none_filter_excludes_a_tagged_file(uploadcare):
+ tag = unique_tag()
+ file_ = upload_image_file(uploadcare, tags=[tag])
+
+ try:
+ wait_until_searchable(
+ uploadcare,
+ FileSearchRequest(tags=TagsFilter(all_=[tag]), sort=BY_NEWEST),
+ )
+
+ response = uploadcare.search_files(
+ FileSearchRequest(tags=TagsFilter(none_=[tag]), sort=BY_NEWEST),
+ limit=5,
+ )
+
+ found = {str(result.uuid) for result in response.results}
+ assert file_.uuid not in found
+ finally:
+ file_.delete()
+
+
+def test_size_range(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(size=SizeRange(gt=1), sort=BY_NEWEST), limit=3
+ )
+
+ assert all(result.size > 1 for result in response.results)
+
+
+def test_sort_descending_by_size(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(is_image=True, sort=["-size"]), limit=5
+ )
+
+ sizes = [result.size for result in response.results]
+ assert sizes == sorted(sizes, reverse=True)
+
+
+def test_include_appdata(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(is_image=True, sort=BY_NEWEST),
+ limit=1,
+ include_appdata=True,
+ )
+
+ if not response.results:
+ pytest.skip("the project has no images to search")
+
+ assert response.results[0].appdata is not None
+
+
+def test_appdata_is_absent_without_the_flag(uploadcare):
+ response = uploadcare.search_files(
+ FileSearchRequest(is_image=True, sort=BY_NEWEST), limit=1
+ )
+
+ if not response.results:
+ pytest.skip("the project has no images to search")
+
+ assert response.results[0].appdata is None
+
+
+def test_pagination_yields_no_duplicates(uploadcare):
+ """Each page is requested at a locally computed offset."""
+ uuids = [
+ str(result.uuid)
+ for result in uploadcare.iterate_search_files(
+ FileSearchRequest(is_image=True, sort=BY_NEWEST),
+ limit=5,
+ request_limit=2,
+ )
+ ]
+
+ assert len(uuids) <= 5
+ assert len(uuids) == len(set(uuids))
+
+
+def test_pagination_matches_a_single_page(uploadcare):
+ request = FileSearchRequest(is_image=True, sort=BY_NEWEST)
+
+ page = uploadcare.search_files(request, limit=4)
+ iterated = list(
+ uploadcare.iterate_search_files(request, limit=4, request_limit=2)
+ )
+
+ assert [str(result.uuid) for result in iterated] == [
+ str(result.uuid) for result in page.results
+ ]
+
+
+def test_dict_request(uploadcare):
+ """A plain dict in the SDK's shape works like the model."""
+ response = uploadcare.search_files(
+ {"is_image": True, "sort": ["-datetime_uploaded"]}, limit=1
+ )
+
+ assert isinstance(response.total, int)
diff --git a/tests/integration/ucare_cli/test_file_tags.py b/tests/integration/ucare_cli/test_file_tags.py
index 70b72eb..8627deb 100644
--- a/tests/integration/ucare_cli/test_file_tags.py
+++ b/tests/integration/ucare_cli/test_file_tags.py
@@ -103,3 +103,15 @@ def test_upload_with_tags(capsys, keys, uploadcare):
assert uploadcare.tags_api.get(file_.uuid) == ["cli-test"]
finally:
file_.delete()
+
+
+def test_search_files(capsys, keys):
+ main(
+ arg_namespace(
+ [*keys, "search_files", "--is_image", "true", "--limit", "1"]
+ )
+ )
+
+ response = _output(capsys)
+ assert isinstance(response["total"], int)
+ assert len(response["results"]) <= 1
diff --git a/tests/integration/utils.py b/tests/integration/utils.py
index 55e7320..c453cba 100644
--- a/tests/integration/utils.py
+++ b/tests/integration/utils.py
@@ -1,6 +1,9 @@
# coding: utf-8
from __future__ import unicode_literals
+import random
+import re
+import time
from pathlib import Path
from tempfile import NamedTemporaryFile
@@ -8,6 +11,13 @@
ASSETS_PATH = Path(__file__).parent / "assets"
IMAGE_PATH = ASSETS_PATH / "img.png"
+# A freshly uploaded file takes about ten seconds to become searchable.
+SEARCH_INDEXING_TIMEOUT = 90
+SEARCH_INDEXING_INTERVAL = 2
+
+# Tags allow Latin letters, digits, `-`, `_` and `.` only.
+TERM_PATTERN = re.compile(r"[A-Za-z0-9]{4,}")
+
def upload_image_file(uploadcare, tags=None):
"""Upload the test image, optionally with tags.
@@ -18,6 +28,49 @@ def upload_image_file(uploadcare, tags=None):
return uploadcare.upload(fh, store=False, tags=tags)
+def unique_tag(prefix="pyuploadcare-test"):
+ """A tag no other test run will collide on."""
+ return f"{prefix}-{random.randint(10 ** 9, 10 ** 10)}"
+
+
+def search_term(filename):
+ """A term from `filename` long enough for a full text condition.
+
+ ``query`` and ``phrase`` values must be at least 4 characters, so a
+ filename without such a run cannot be searched for. Returns ``None`` then.
+ """
+ match = TERM_PATTERN.search(filename or "")
+ return match.group(0) if match else None
+
+
+def wait_until_searchable(
+ uploadcare,
+ request,
+ limit=5,
+ timeout=SEARCH_INDEXING_TIMEOUT,
+ interval=SEARCH_INDEXING_INTERVAL,
+):
+ """Poll search until `request` returns results, then return the response.
+
+ Search indexing is asynchronous, so a file is not findable the moment it
+ is uploaded.
+ """
+ deadline = time.monotonic() + timeout
+
+ while True:
+ response = uploadcare.search_files(request, limit=limit)
+
+ if response.results:
+ return response
+
+ if time.monotonic() >= deadline:
+ raise AssertionError(
+ f"search returned nothing within {timeout}s for {request!r}"
+ )
+
+ time.sleep(interval)
+
+
def upload_tmp_txt_file(uploadcare, content=""):
tmp_txt_file = NamedTemporaryFile(mode="wb", delete=False)
tmp_txt_file.write(content.encode("utf-8"))