Encode URL user-inputted string to prevent injection

src/typesense/alias.py

@@ -1,11 +1,15 @@
+from .utils import encodeURIComponent
+
+
 class Alias(object):
     def __init__(self, api_call, name):
         self.api_call = api_call
         self.name = name
 
     def _endpoint_path(self):
         from .aliases import Aliases
-        return u"{0}/{1}".format(Aliases.RESOURCE_PATH, self.name)
+
+        return "{0}/{1}".format(Aliases.RESOURCE_PATH, encodeURIComponent(self.name))
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/aliases.py

@@ -1,8 +1,10 @@
 from typesense.alias import Alias
 
+from .utils import encodeURIComponent
+
 
 class Aliases(object):
-    RESOURCE_PATH = '/aliases'
+    RESOURCE_PATH = "/aliases"
 
     def __init__(self, api_call):
         self.api_call = api_call
@@ -15,7 +17,7 @@ def __getitem__(self, name):
         return self.aliases.get(name)
 
     def _endpoint_path(self, alias_name):
-        return u"{0}/{1}".format(Aliases.RESOURCE_PATH, alias_name)
+        return "{0}/{1}".format(Aliases.RESOURCE_PATH, encodeURIComponent(alias_name))
 
     def upsert(self, name, mapping):
         return self.api_call.put(self._endpoint_path(name), mapping)

src/typesense/analytics_rule.py

@@ -1,11 +1,17 @@
+from .utils import encodeURIComponent
+
+
 class AnalyticsRule(object):
     def __init__(self, api_call, rule_id):
         self.api_call = api_call
         self.rule_id = rule_id
 
     def _endpoint_path(self):
         from .analytics_rules import AnalyticsRules
-        return u"{0}/{1}".format(AnalyticsRules.RESOURCE_PATH, self.rule_id)
+
+        return "{0}/{1}".format(
+            AnalyticsRules.RESOURCE_PATH, encodeURIComponent(self.rule_id)
+        )
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/analytics_rules.py

@@ -1,8 +1,9 @@
 from .analytics_rule import AnalyticsRule
+from .utils import encodeURIComponent
 
 
 class AnalyticsRules(object):
-    RESOURCE_PATH = '/analytics/rules'
+    RESOURCE_PATH = "/analytics/rules"
 
     def __init__(self, api_call):
         self.api_call = api_call
@@ -19,8 +20,9 @@ def create(self, rule, params=None):
         return self.api_call.post(AnalyticsRules.RESOURCE_PATH, rule, params)
 
     def upsert(self, id, rule):
-        return self.api_call.put(u"{0}/{1}".format(AnalyticsRules.RESOURCE_PATH, id), rule)
+        return self.api_call.put(
+            "{0}/{1}".format(AnalyticsRules.RESOURCE_PATH, encodeURIComponent(id)), rule
+        )
 
     def retrieve(self):
         return self.api_call.get(AnalyticsRules.RESOURCE_PATH)
-

src/typesense/collection.py

@@ -1,6 +1,7 @@
+from .documents import Documents
 from .overrides import Overrides
 from .synonyms import Synonyms
-from .documents import Documents
+from .utils import encodeURIComponent
 
 
 class Collection(object):
@@ -13,7 +14,10 @@ def __init__(self, api_call, name):
 
     def _endpoint_path(self):
         from .collections import Collections
-        return u"{0}/{1}".format(Collections.RESOURCE_PATH, self.name)
+
+        return "{0}/{1}".format(
+            Collections.RESOURCE_PATH, encodeURIComponent(self.name)
+        )
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/conversation_model.py

@@ -1,11 +1,17 @@
+from .utils import encodeURIComponent
+
+
 class ConversationModel(object):
     def __init__(self, api_call, model_id):
         self.model_id = model_id
         self.api_call = api_call
 
     def _endpoint_path(self):
         from .conversations_models import ConversationsModels
-        return u"{0}/{1}".format(ConversationsModels.RESOURCE_PATH, self.model_id)
+
+        return "{0}/{1}".format(
+            ConversationsModels.RESOURCE_PATH, encodeURIComponent(self.model_id)
+        )
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/document.py

@@ -1,14 +1,22 @@
+from .utils import encodeURIComponent
+
+
 class Document(object):
     def __init__(self, api_call, collection_name, document_id):
         self.api_call = api_call
         self.collection_name = collection_name
         self.document_id = document_id
 
     def _endpoint_path(self):
-        from .documents import Documents
         from .collections import Collections
-        return u"{0}/{1}/{2}/{3}".format(Collections.RESOURCE_PATH, self.collection_name, Documents.RESOURCE_PATH,
-                                         self.document_id)
+        from .documents import Documents
+
+        return "{0}/{1}/{2}/{3}".format(
+            Collections.RESOURCE_PATH,
+            encodeURIComponent(self.collection_name),
+            Documents.RESOURCE_PATH,
+            encodeURIComponent(self.document_id),
+        )
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/documents.py

@@ -1,15 +1,17 @@
 import json
+from collections.abc import Iterable
 
 from typesense.exceptions import TypesenseClientError
 
 from .document import Document
 from .logger import logger
-from .validation import validate_search
 from .preprocess import stringify_search_params
-from collections.abc import Iterable
+from .utils import encodeURIComponent
+from .validation import validate_search
+
 
 class Documents(object):
-    RESOURCE_PATH = 'documents'
+    RESOURCE_PATH = "documents"
 
     def __init__(self, api_call, collection_name):
         self.api_call = api_call
@@ -18,38 +20,44 @@ def __init__(self, api_call, collection_name):
 
     def __getitem__(self, document_id):
         if document_id not in self.documents:
-            self.documents[document_id] = Document(self.api_call, self.collection_name, document_id)
+            self.documents[document_id] = Document(
+                self.api_call, self.collection_name, document_id
+            )
 
         return self.documents[document_id]
 
     def _endpoint_path(self, action=None):
         from .collections import Collections
 
-        action = action or ''
-        return u"{0}/{1}/{2}/{3}".format(Collections.RESOURCE_PATH, self.collection_name, Documents.RESOURCE_PATH,
-                                         action)
+        action = action or ""
+        return "{0}/{1}/{2}/{3}".format(
+            Collections.RESOURCE_PATH,
+            encodeURIComponent(self.collection_name),
+            Documents.RESOURCE_PATH,
+            action,
+        )
 
     def create(self, document, params=None):
         params = params or {}
-        params['action'] = 'create'
+        params["action"] = "create"
         return self.api_call.post(self._endpoint_path(), document, params)
 
     def create_many(self, documents, params=None):
-        logger.warning('`create_many` is deprecated: please use `import_`.')
+        logger.warning("`create_many` is deprecated: please use `import_`.")
         return self.import_(documents, params)
 
     def upsert(self, document, params=None):
         params = params or {}
-        params['action'] = 'upsert'
+        params["action"] = "upsert"
         return self.api_call.post(self._endpoint_path(), document, params)
 
     def update(self, document, params=None):
         params = params or {}
-        params['action'] = 'update'
+        params["action"] = "update"
         return self.api_call.patch(self._endpoint_path(), document, params)
 
     def import_jsonl(self, documents_jsonl):
-        logger.warning('`import_jsonl` is deprecated: please use `import_`.')
+        logger.warning("`import_jsonl` is deprecated: please use `import_`.")
         return self.import_(documents_jsonl)
 
     # `documents` can be either a list of document objects (or)
@@ -61,7 +69,7 @@ def import_(self, documents, params=None, batch_size=None):
                 batch = []
                 for document in documents:
                     batch.append(document)
-                    if (len(batch) == batch_size):
+                    if len(batch) == batch_size:
                         api_response = self.import_(batch, params)
                         response_objs.extend(api_response)
                         batch = []
@@ -75,33 +83,45 @@ def import_(self, documents, params=None, batch_size=None):
                     document_strs.append(json.dumps(document))
 
                 if len(document_strs) == 0:
-                    raise TypesenseClientError(f"Cannot import an empty list of documents.")
+                    raise TypesenseClientError(
+                        "Cannot import an empty list of documents."
+                    )
 
-                docs_import = '\n'.join(document_strs)
-                api_response = self.api_call.post(self._endpoint_path('import'), docs_import, params, as_json=False)
-                res_obj_strs = api_response.split('\n')
+                docs_import = "\n".join(document_strs)
+                api_response = self.api_call.post(
+                    self._endpoint_path("import"), docs_import, params, as_json=False
+                )
+                res_obj_strs = api_response.split("\n")
 
                 response_objs = []
                 for res_obj_str in res_obj_strs:
                     try:
                         res_obj_json = json.loads(res_obj_str)
                     except json.JSONDecodeError as e:
-                        raise TypesenseClientError(f"Invalid response - {res_obj_str}") from e
+                        raise TypesenseClientError(
+                            f"Invalid response - {res_obj_str}"
+                        ) from e
                     response_objs.append(res_obj_json)
 
             return response_objs
         else:
-            api_response = self.api_call.post(self._endpoint_path('import'), documents, params, as_json=False)
+            api_response = self.api_call.post(
+                self._endpoint_path("import"), documents, params, as_json=False
+            )
             return api_response
 
     def export(self, params=None):
-        api_response = self.api_call.get(self._endpoint_path('export'), params, as_json=False)
+        api_response = self.api_call.get(
+            self._endpoint_path("export"), params, as_json=False
+        )
         return api_response
 
     def search(self, search_parameters):
         stringified_search_params = stringify_search_params(search_parameters)
         validate_search(stringified_search_params)
-        return self.api_call.get(self._endpoint_path('search'), stringified_search_params)
+        return self.api_call.get(
+            self._endpoint_path("search"), stringified_search_params
+        )
 
     def delete(self, params=None):
         return self.api_call.delete(self._endpoint_path(), params)

src/typesense/key.py

@@ -1,3 +1,4 @@
+from .utils import encodeURIComponent
 
 
 class Key(object):
@@ -7,7 +8,8 @@ def __init__(self, api_call, key_id):
 
     def _endpoint_path(self):
         from .keys import Keys
-        return u"{0}/{1}".format(Keys.RESOURCE_PATH, self.key_id)
+
+        return "{0}/{1}".format(Keys.RESOURCE_PATH, encodeURIComponent(self.key_id))
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/override.py

@@ -1,14 +1,22 @@
+from .utils import encodeURIComponent
+
+
 class Override(object):
     def __init__(self, api_call, collection_name, override_id):
         self.api_call = api_call
         self.collection_name = collection_name
         self.override_id = override_id
 
     def _endpoint_path(self):
-        from .overrides import Overrides
         from .collections import Collections
-        return u"{0}/{1}/{2}/{3}".format(Collections.RESOURCE_PATH, self.collection_name, Overrides.RESOURCE_PATH,
-                                         self.override_id)
+        from .overrides import Overrides
+
+        return "{0}/{1}/{2}/{3}".format(
+            Collections.RESOURCE_PATH,
+            encodeURIComponent(self.collection_name),
+            Overrides.RESOURCE_PATH,
+            encodeURIComponent(self.override_id),
+        )
 
     def retrieve(self):
         return self.api_call.get(self._endpoint_path())

src/typesense/overrides.py

@@ -1,8 +1,9 @@
 from .override import Override
+from .utils import encodeURIComponent
 
 
 class Overrides(object):
-    RESOURCE_PATH = 'overrides'
+    RESOURCE_PATH = "overrides"
 
     def __init__(self, api_call, collection_name):
         self.api_call = api_call
@@ -11,15 +12,22 @@ def __init__(self, api_call, collection_name):
 
     def __getitem__(self, override_id):
         if override_id not in self.overrides:
-            self.overrides[override_id] = Override(self.api_call, self.collection_name, override_id)
+            self.overrides[override_id] = Override(
+                self.api_call, self.collection_name, override_id
+            )
 
         return self.overrides[override_id]
 
     def _endpoint_path(self, override_id=None):
         from .collections import Collections
-        override_id = override_id or ''
-        return u"{0}/{1}/{2}/{3}".format(Collections.RESOURCE_PATH, self.collection_name,
-                                         Overrides.RESOURCE_PATH, override_id)
+
+        override_id = override_id or ""
+        return "{0}/{1}/{2}/{3}".format(
+            Collections.RESOURCE_PATH,
+            encodeURIComponent(self.collection_name),
+            Overrides.RESOURCE_PATH,
+            encodeURIComponent(override_id),
+        )
 
     def upsert(self, id, schema):
         return self.api_call.put(self._endpoint_path(id), schema)

src/typesense/stopwords.py

@@ -1,21 +1,29 @@
 from .stopwords_set import StopwordsSet
+from .utils import encodeURIComponent
 
 
 class Stopwords(object):
-    RESOURCE_PATH = '/stopwords'
+    RESOURCE_PATH = "/stopwords"
 
     def __init__(self, api_call):
         self.api_call = api_call
         self.stopwords_sets = {}
 
     def __getitem__(self, stopwords_set_id):
         if stopwords_set_id not in self.stopwords_sets:
-            self.stopwords_sets[stopwords_set_id] = StopwordsSet(self.api_call, stopwords_set_id)
+            self.stopwords_sets[stopwords_set_id] = StopwordsSet(
+                self.api_call, stopwords_set_id
+            )
 
         return self.stopwords_sets.get(stopwords_set_id)
 
     def upsert(self, stopwords_set_id, stopwords_set):
-        return self.api_call.put('{}/{}'.format(Stopwords.RESOURCE_PATH, stopwords_set_id), stopwords_set)
+        return self.api_call.put(
+            "{}/{}".format(
+                Stopwords.RESOURCE_PATH, encodeURIComponent(stopwords_set_id)
+            ),
+            stopwords_set,
+        )
 
     def retrieve(self):
-        return self.api_call.get('{0}'.format(Stopwords.RESOURCE_PATH))
+        return self.api_call.get("{0}".format(Stopwords.RESOURCE_PATH))