Merge branch 'main' into dependabot/pip/setuptools-gte-69.2-and-lt-73.0

aredis_om/model/encoders.py

@@ -90,7 +90,7 @@ def jsonable_encoder(
             sqlalchemy_safe=sqlalchemy_safe,
         )
     if dataclasses.is_dataclass(obj):
-        return dataclasses.asdict(obj)
+        return dataclasses.asdict(obj)  # type: ignore[call-overload]
     if isinstance(obj, Enum):
         return obj.value
     if isinstance(obj, PurePath):

aredis_om/model/model.py

@@ -14,6 +14,7 @@
     ClassVar,
     Dict,
     List,
+    Literal,
     Mapping,
     Optional,
     Sequence,
@@ -141,10 +142,10 @@ def embedded(cls):
 
 def is_supported_container_type(typ: Optional[type]) -> bool:
     # TODO: Wait, why don't we support indexing sets?
-    if typ == list or typ == tuple:
+    if typ == list or typ == tuple or typ == Literal:
         return True
     unwrapped = get_origin(typ)
-    return unwrapped == list or unwrapped == tuple
+    return unwrapped == list or unwrapped == tuple or unwrapped == Literal
 
 
 def validate_model_fields(model: Type["RedisModel"], field_values: Dict[str, Any]):
@@ -872,7 +873,9 @@ def resolve_redisearch_query(cls, expression: ExpressionOrNegated) -> str:
 
         return result
 
-    async def execute(self, exhaust_results=True, return_raw_result=False):
+    async def execute(
+        self, exhaust_results=True, return_raw_result=False, return_query_args=False
+    ):
         args: List[Union[str, bytes]] = [
             "FT.SEARCH",
             self.model.Meta.index_name,
@@ -897,6 +900,9 @@ async def execute(self, exhaust_results=True, return_raw_result=False):
         if self.nocontent:
             args.append("NOCONTENT")
 
+        if return_query_args:
+            return self.model.Meta.index_name, args
+
         # Reset the cache if we're executing from offset 0.
         if self.offset == 0:
             self._model_cache.clear()
@@ -930,6 +936,10 @@ async def execute(self, exhaust_results=True, return_raw_result=False):
             self._model_cache += _results
         return self._model_cache
 
+    async def get_query(self):
+        query = self.copy()
+        return await query.execute(return_query_args=True)
+
     async def first(self):
         query = self.copy(offset=0, limit=1, sort_fields=self.sort_fields)
         results = await query.execute(exhaust_results=False)
@@ -1414,6 +1424,8 @@ def outer_type_or_annotation(field):
         if not isinstance(field.annotation, type):
             raise AttributeError(f"could not extract outer type from field {field}")
         return field.annotation
+    elif get_origin(field.annotation) == Literal:
+        return str
     else:
         return field.annotation.__args__[0]
 
@@ -2057,21 +2069,33 @@ def schema_for_type(
         # find any values marked as indexed.
         if is_container_type and not is_vector:
             field_type = get_origin(typ)
-            embedded_cls = get_args(typ)
-            if not embedded_cls:
-                log.warning(
-                    "Model %s defined an empty list or tuple field: %s", cls, name
+            if field_type == Literal:
+                path = f"{json_path}.{name}"
+                return cls.schema_for_type(
+                    path,
+                    name,
+                    name_prefix,
+                    str,
+                    field_info,
+                    parent_type=field_type,
+                )
+            else:
+                embedded_cls = get_args(typ)
+                if not embedded_cls:
+                    log.warning(
+                        "Model %s defined an empty list or tuple field: %s", cls, name
+                    )
+                    return ""
+                path = f"{json_path}.{name}[*]"
+                embedded_cls = embedded_cls[0]
+                return cls.schema_for_type(
+                    path,
+                    name,
+                    name_prefix,
+                    embedded_cls,
+                    field_info,
+                    parent_type=field_type,
                 )
-                return ""
-            embedded_cls = embedded_cls[0]
-            return cls.schema_for_type(
-                f"{json_path}.{name}[*]",
-                name,
-                name_prefix,
-                embedded_cls,
-                field_info,
-                parent_type=field_type,
-            )
         elif field_is_model:
             name_prefix = f"{name_prefix}_{name}" if name_prefix else name
             sub_fields = []