From 911dd49419e32ba20df028d53d4ec2386a0dddb0 Mon Sep 17 00:00:00 2001 From: coffeepy Date: Wed, 4 Jun 2025 13:02:59 -0500 Subject: [PATCH] feat: add search query params and tests --- pythonik/specs/search.py | 59 ++++++++++++--- pythonik/tests/test_search.py | 131 +++++++++++++++++++++++++++------- 2 files changed, 154 insertions(+), 36 deletions(-) diff --git a/pythonik/specs/search.py b/pythonik/specs/search.py index d5212ba..21aa1d6 100644 --- a/pythonik/specs/search.py +++ b/pythonik/specs/search.py @@ -1,4 +1,4 @@ -from typing import Union, Dict, Any +from typing import Union, Dict, Any, Optional from pythonik.models.base import Response from pythonik.models.search.search_body import SearchBody @@ -15,24 +15,63 @@ class SearchSpec(Spec): def search( self, search_body: Union[SearchBody, Dict[str, Any]], + per_page: Optional[int] = None, + page: Optional[int] = None, + scroll: Optional[bool] = None, # Deprecated + scroll_id: Optional[str] = None, # Deprecated + generate_signed_url: Optional[bool] = None, + generate_signed_download_url: Optional[bool] = None, + generate_signed_proxy_url: Optional[bool] = None, + save_search_history: Optional[bool] = None, exclude_defaults: bool = True, - **kwargs + **kwargs, ) -> Response: # Response.data will be SearchResponse """ - Search iconik + Search iconik. + Corresponds to POST /v1/search/ Args: - search_body: Search parameters, either as SearchBody model or dict - exclude_defaults: Whether to exclude default values when dumping Pydantic models - **kwargs: Additional kwargs to pass to the request + search_body: Search parameters, either as SearchBody model or dict. + per_page: The number of documents for each page. + page: Which page number to fetch. + scroll: If true, uses scroll pagination. (Deprecated, use search_after in body). + scroll_id: Scroll ID for scroll pagination. (Deprecated). + generate_signed_url: Set to false if you don't need a URL, will speed things up. + generate_signed_download_url: Set to true if you also want the file download URLs generated. + generate_signed_proxy_url: Set to true if you want to generate signed download urls for proxies. + save_search_history: Set to false if you don't want to save the search to the history. + exclude_defaults: Whether to exclude default values when dumping Pydantic models for the request body. + **kwargs: Additional kwargs to pass to the request (e.g., headers). Returns: - Response with SearchResponse data model + Response with SearchResponse data model. """ - json_data = self._prepare_model_data(search_body, exclude_defaults=exclude_defaults) + json_data = self._prepare_model_data( + search_body, exclude_defaults=exclude_defaults + ) + + params = {} + if per_page is not None: + params["per_page"] = per_page + if page is not None: + params["page"] = page + if scroll is not None: + params["scroll"] = scroll + if scroll_id is not None: + params["scroll_id"] = scroll_id + if generate_signed_url is not None: + params["generate_signed_url"] = generate_signed_url + if generate_signed_download_url is not None: + params["generate_signed_download_url"] = generate_signed_download_url + if generate_signed_proxy_url is not None: + params["generate_signed_proxy_url"] = generate_signed_proxy_url + if save_search_history is not None: + params["save_search_history"] = save_search_history + resp = self._post( - SEARCH_PATH, + SEARCH_PATH, # Use the new path constant, which is "" json=json_data, - **kwargs + params=params if params else None, + **kwargs, ) return self.parse_response(resp, SearchResponse) diff --git a/pythonik/tests/test_search.py b/pythonik/tests/test_search.py index 4ab3bed..5354a83 100644 --- a/pythonik/tests/test_search.py +++ b/pythonik/tests/test_search.py @@ -1,44 +1,123 @@ import uuid +import pytest import requests_mock -from pythonik.client import PythonikClient +# from urllib.parse import parse_qs # Unused import removed -from pythonik.models.metadata.views import ViewMetadata -from pythonik.models.mutation.metadata.mutate import ( - UpdateMetadata, - UpdateMetadataResponse, -) +from pythonik.client import PythonikClient from pythonik.models.search.search_body import Filter, SearchBody, SortItem, Term -from pythonik.specs.metadata import ( - ASSET_METADATA_FROM_VIEW_PATH, - UPDATE_ASSET_METADATA, - MetadataSpec, -) from pythonik.specs.search import SEARCH_PATH, SearchSpec +# Unused imports removed by Cascade: +# from pythonik.models.metadata.views import ViewMetadata +# from pythonik.models.mutation.metadata.mutate import ( +# UpdateMetadata, +# UpdateMetadataResponse, +# ) +# from pythonik.specs.metadata import ( +# ASSET_METADATA_FROM_VIEW_PATH, +# UPDATE_ASSET_METADATA, +# MetadataSpec, +# ) -def test_search_assets(): +def test_search_assets_basic(): + """Test basic search functionality, similar to original test but using named params.""" with requests_mock.Mocker() as m: app_id = str(uuid.uuid4()) auth_token = str(uuid.uuid4()) asset_id = str(uuid.uuid4()) - view_id = str(uuid.uuid4()) + # view_id = str(uuid.uuid4()) # Unused variable removed - # needs model - params = {"generate_signed_url": "true", "generate_signed_download_url": "true"} + search_criteria = SearchBody( + doc_types=["assets"], + query=f"id:{asset_id}", + filter=Filter(operator="AND", terms=[Term(name="status", value="active")]), + sort=[SortItem(name="date_created", order="desc")] + ) - # search criteria - search_chriteria = SearchBody() - search_chriteria.doc_types = ["assets"] - search_chriteria.query = f"id:{asset_id}" + mock_address = SearchSpec.gen_url(SEARCH_PATH) + # Mock will match the base address; query params will be checked on m.last_request.qs + matcher = m.post(mock_address, json=search_criteria.model_dump()) - search_chriteria.filter = Filter( - operator="AND", terms=[Term(name="status", value="active")] + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + client.search().search( + search_body=search_criteria, + generate_signed_url=True, + generate_signed_download_url=True ) - # get only active assets + assert matcher.called_once + expected_qs = { + 'generate_signed_url': ['true'], + 'generate_signed_download_url': ['true'] + } + assert m.last_request.qs == expected_qs - search_chriteria.sort = [SortItem(name="date_created", order="desc")] - mock_address = SearchSpec.gen_url(SEARCH_PATH) - m.post(mock_address, json=search_chriteria.model_dump()) +# Test cases for various query parameter combinations +# Each tuple: (test_id, query_params_for_search_method, expected_query_string_dict) +search_param_test_cases = [ + ( + "pagination", + {"per_page": 20, "page": 3}, + {"per_page": ["20"], "page": ["3"]} + ), + ( + "signed_urls_off_and_proxy_on", + {"generate_signed_url": False, "generate_signed_download_url": False, "generate_signed_proxy_url": True}, + {"generate_signed_url": ["false"], "generate_signed_download_url": ["false"], "generate_signed_proxy_url": ["true"]} + ), + ( + "save_history_off", + {"save_search_history": False}, + {"save_search_history": ["false"]} + ), + ( + "scroll_params_active", + {"scroll": True, "scroll_id": "test_scroll_123"}, + {"scroll": ["true"], "scroll_id": ["test_scroll_123"]} + ), + ( + "all_bools_mixed_values", + { + "generate_signed_url": True, + "generate_signed_download_url": False, + "generate_signed_proxy_url": True, + "save_search_history": False, + }, + { + "generate_signed_url": ["true"], + "generate_signed_download_url": ["false"], + "generate_signed_proxy_url": ["true"], + "save_search_history": ["false"], + } + ), + ( + "no_extra_query_params", + {}, + {} + ), +] + +@pytest.mark.parametrize("test_id, query_params, expected_qs_dict", search_param_test_cases) +def test_search_with_various_query_params(test_id, query_params, expected_qs_dict): + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + + search_body_data = SearchBody( + doc_types=["collections"], + query=f"title:Test Collection AND id:{asset_id}" + ) + + base_mock_address = SearchSpec.gen_url(SEARCH_PATH) + # Mock the base address, query parameters will be checked via m.last_request.qs + matcher = m.post(base_mock_address, json=search_body_data.model_dump()) + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) - client.search().search(search_chriteria, params=params) + client.search().search( + search_body=search_body_data, + **query_params # Pass the dictionary of query params as keyword arguments + ) + + assert matcher.called_once + assert m.last_request.qs == expected_qs_dict