resolve conflict

Signed-off-by: wiseaidev <[email protected]>

Author: wiseaidev

aredis_om/__init__.py

@@ -1,3 +1,4 @@
+from .async_redis import redis  # isort:skip
 from .checks import has_redis_json, has_redisearch
 from .connections import get_redis_connection
 from .model.migrations.migrator import MigrationError, Migrator

aredis_om/async_redis.py

@@ -0,0 +1 @@
+from redis import asyncio as redis

aredis_om/connections.py

@@ -1,19 +1,19 @@
 import os
 
-import aioredis
+from . import redis
 
 
 URL = os.environ.get("REDIS_OM_URL", None)
 
 
-def get_redis_connection(**kwargs) -> aioredis.Redis:
+def get_redis_connection(**kwargs) -> redis.Redis:
     # If someone passed in a 'url' parameter, or specified a REDIS_OM_URL
     # environment variable, we'll create the Redis client from the URL.
     url = kwargs.pop("url", URL)
     if url:
-        return aioredis.Redis.from_url(url, **kwargs)
+        return redis.Redis.from_url(url, **kwargs)
 
     # Decode from UTF-8 by default
     if "decode_responses" not in kwargs:
         kwargs["decode_responses"] = True
-    return aioredis.Redis(**kwargs)
+    return redis.Redis(**kwargs)

aredis_om/model/migrations/migrator.py

@@ -4,7 +4,7 @@
 from enum import Enum
 from typing import List, Optional
 
-from aioredis import Redis, ResponseError
+from ... import redis
 
 
 log = logging.getLogger(__name__)
@@ -39,18 +39,19 @@ def schema_hash_key(index_name):
     return f"{index_name}:hash"
 
 
-async def create_index(redis: Redis, index_name, schema, current_hash):
-    db_number = redis.connection_pool.connection_kwargs.get("db")
+async def create_index(conn: redis.Redis, index_name, schema, current_hash):
+    db_number = conn.connection_pool.connection_kwargs.get("db")
     if db_number and db_number > 0:
         raise MigrationError(
             "Creating search indexes is only supported in database 0. "
             f"You attempted to create an index in database {db_number}"
         )
     try:
-        await redis.execute_command(f"ft.info {index_name}")
-    except ResponseError:
-        await redis.execute_command(f"ft.create {index_name} {schema}")
-        await redis.set(schema_hash_key(index_name), current_hash)
+        await conn.execute_command(f"ft.info {index_name}")
+    except redis.ResponseError:
+        await conn.execute_command(f"ft.create {index_name} {schema}")
+        # TODO: remove "type: ignore" when type stubs will be fixed
+        await conn.set(schema_hash_key(index_name), current_hash)  # type: ignore
     else:
         log.info("Index already exists, skipping. Index hash: %s", index_name)
 
@@ -67,7 +68,7 @@ class IndexMigration:
     schema: str
     hash: str
     action: MigrationAction
-    redis: Redis
+    conn: redis.Redis
     previous_hash: Optional[str] = None
 
     async def run(self):
@@ -78,14 +79,14 @@ async def run(self):
 
     async def create(self):
         try:
-            await create_index(self.redis, self.index_name, self.schema, self.hash)
-        except ResponseError:
+            await create_index(self.conn, self.index_name, self.schema, self.hash)
+        except redis.ResponseError:
             log.info("Index already exists: %s", self.index_name)
 
     async def drop(self):
         try:
-            await self.redis.execute_command(f"FT.DROPINDEX {self.index_name}")
-        except ResponseError:
+            await self.conn.execute_command(f"FT.DROPINDEX {self.index_name}")
+        except redis.ResponseError:
             log.info("Index does not exist: %s", self.index_name)
 
 
@@ -105,7 +106,7 @@ async def detect_migrations(self):
 
         for name, cls in model_registry.items():
             hash_key = schema_hash_key(cls.Meta.index_name)
-            redis = cls.db()
+            conn = cls.db()
             try:
                 schema = cls.redisearch_schema()
             except NotImplementedError:
@@ -114,21 +115,21 @@ async def detect_migrations(self):
             current_hash = hashlib.sha1(schema.encode("utf-8")).hexdigest()  # nosec
 
             try:
-                await redis.execute_command("ft.info", cls.Meta.index_name)
-            except ResponseError:
+                await conn.execute_command("ft.info", cls.Meta.index_name)
+            except redis.ResponseError:
                 self.migrations.append(
                     IndexMigration(
                         name,
                         cls.Meta.index_name,
                         schema,
                         current_hash,
                         MigrationAction.CREATE,
-                        redis,
+                        conn,
                     )
                 )
                 continue
 
-            stored_hash = await redis.get(hash_key)
+            stored_hash = await conn.get(hash_key)
             schema_out_of_date = current_hash != stored_hash
 
             if schema_out_of_date:
@@ -140,7 +141,7 @@ async def detect_migrations(self):
                         schema,
                         current_hash,
                         MigrationAction.DROP,
-                        redis,
+                        conn,
                         stored_hash,
                     )
                 )
@@ -151,7 +152,7 @@ async def detect_migrations(self):
                         schema,
                         current_hash,
                         MigrationAction.CREATE,
-                        redis,
+                        conn,
                         stored_hash,
                     )
                 )

aredis_om/model/model.py

@@ -24,8 +24,7 @@
     no_type_check,
 )
 
-import aioredis
-from aioredis.client import Pipeline
+from more_itertools import ichunked
 from pydantic import BaseModel, validator
 from pydantic.fields import FieldInfo as PydanticFieldInfo
 from pydantic.fields import ModelField, Undefined, UndefinedType
@@ -35,9 +34,10 @@
 from typing_extensions import Protocol, get_args, get_origin
 from ulid import ULID
 
+from .. import redis
 from ..checks import has_redis_json, has_redisearch
 from ..connections import get_redis_connection
-from ..unasync_util import ASYNC_MODE
+from ..util import ASYNC_MODE
 from .encoders import jsonable_encoder
 from .render_tree import render_tree
 from .token_escaper import TokenEscaper
@@ -760,6 +760,9 @@ async def all(self, batch_size=DEFAULT_PAGE_SIZE):
             return await query.execute()
         return await self.execute()
 
+    async def page(self, offset=0, limit=10):
+        return await self.copy(offset=offset, limit=limit).execute()
+
     def sort_by(self, *fields: str):
         if not fields:
             return self
@@ -975,7 +978,7 @@ class BaseMeta(Protocol):
     global_key_prefix: str
     model_key_prefix: str
     primary_key_pattern: str
-    database: aioredis.Redis
+    database: redis.Redis
     primary_key: PrimaryKey
     primary_key_creator_cls: Type[PrimaryKeyCreator]
     index_name: str
@@ -994,7 +997,7 @@ class DefaultMeta:
     global_key_prefix: Optional[str] = None
     model_key_prefix: Optional[str] = None
     primary_key_pattern: Optional[str] = None
-    database: Optional[aioredis.Redis] = None
+    database: Optional[redis.Redis] = None
     primary_key: Optional[PrimaryKey] = None
     primary_key_creator_cls: Optional[Type[PrimaryKeyCreator]] = None
     index_name: Optional[str] = None
@@ -1115,9 +1118,17 @@ def key(self):
         return self.make_primary_key(pk)
 
     @classmethod
-    async def delete(cls, pk: Any) -> int:
+    async def _delete(cls, db, *pks):
+        return await db.delete(*pks)
+
+    @classmethod
+    async def delete(
+        cls, pk: Any, pipeline: Optional[redis.client.Pipeline] = None
+    ) -> int:
         """Delete data at this key."""
-        return await cls.db().delete(cls.make_primary_key(pk))
+        db = cls._get_db(pipeline)
+
+        return await cls._delete(db, cls.make_primary_key(pk))
 
     @classmethod
     async def get(cls, pk: Any) -> "RedisModel":
@@ -1127,14 +1138,15 @@ async def update(self, **field_values):
         """Update this model instance with the specified key-value pairs."""
         raise NotImplementedError
 
-    async def save(self, pipeline: Optional[Pipeline] = None) -> "RedisModel":
+    async def save(
+        self, pipeline: Optional[redis.client.Pipeline] = None
+    ) -> "RedisModel":
         raise NotImplementedError
 
-    async def expire(self, num_seconds: int, pipeline: Optional[Pipeline] = None):
-        if pipeline is None:
-            db = self.db()
-        else:
-            db = pipeline
+    async def expire(
+        self, num_seconds: int, pipeline: Optional[redis.client.Pipeline] = None
+    ):
+        db = self._get_db(pipeline)
 
         # TODO: Wrap any Redis response errors in a custom exception?
         await db.expire(self.make_primary_key(self.pk), num_seconds)
@@ -1223,19 +1235,10 @@ def get_annotations(cls):
     async def add(
         cls,
         models: Sequence["RedisModel"],
-        pipeline: Optional[Pipeline] = None,
+        pipeline: Optional[redis.client.Pipeline] = None,
         pipeline_verifier: Callable[..., Any] = verify_pipeline_response,
     ) -> Sequence["RedisModel"]:
-        if pipeline is None:
-            # By default, send commands in a pipeline. Saving each model will
-            # be atomic, but Redis may process other commands in between
-            # these saves.
-            db = cls.db().pipeline(transaction=False)
-        else:
-            # If the user gave us a pipeline, add our commands to that. The user
-            # will be responsible for executing the pipeline after they've accumulated
-            # the commands they want to send.
-            db = pipeline
+        db = cls._get_db(pipeline, bulk=True)
 
         for model in models:
             # save() just returns the model, we don't need that here.
@@ -1249,6 +1252,31 @@ async def add(
 
         return models
 
+    @classmethod
+    def _get_db(
+        self, pipeline: Optional[redis.client.Pipeline] = None, bulk: bool = False
+    ):
+        if pipeline is not None:
+            return pipeline
+        elif bulk:
+            return self.db().pipeline(transaction=False)
+        else:
+            return self.db()
+
+    @classmethod
+    async def delete_many(
+        cls,
+        models: Sequence["RedisModel"],
+        pipeline: Optional[redis.client.Pipeline] = None,
+    ) -> int:
+        db = cls._get_db(pipeline)
+
+        for chunk in ichunked(models, 100):
+            pks = [cls.make_primary_key(model.pk) for model in chunk]
+            await cls._delete(db, *pks)
+
+        return len(models)
+
     @classmethod
     def redisearch_schema(cls):
         raise NotImplementedError
@@ -1283,11 +1311,11 @@ def __init_subclass__(cls, **kwargs):
                     f"HashModels cannot index dataclass fields. Field: {name}"
                 )
 
-    def dict(self) -> Dict[str, Any]:
+    def dict(self) -> Dict[str, Any]:  # type: ignore
         # restore none values
         return dict(self)
 
-    async def save(self, pipeline: Optional[Pipeline] = None) -> "HashModel":
+    async def save(self, pipeline: Optional[redis.client.Pipeline] = None) -> "HashModel":
         self.check()
         if pipeline is None:
             db = self.db()
@@ -1461,12 +1489,12 @@ def __init__(self, *args, **kwargs):
             )
         super().__init__(*args, **kwargs)
 
-    async def save(self, pipeline: Optional[Pipeline] = None) -> "JsonModel":
+    async def save(
+        self, pipeline: Optional[redis.client.Pipeline] = None
+    ) -> "JsonModel":
         self.check()
-        if pipeline is None:
-            db = self.db()
-        else:
-            db = pipeline
+        db = self._get_db(pipeline)
+
         # TODO: Wrap response errors in a custom exception?
         await db.execute_command("JSON.SET", self.key(), ".", self.json())
         return self

aredis_om/sync_redis.py

@@ -0,0 +1 @@
+import redis

aredis_om/unasync_util.py

@@ -0,0 +1,12 @@
+import inspect
+
+
+def is_async_mode():
+    async def f():
+        """Unasync transforms async functions in sync functions"""
+        return None
+
+    return inspect.iscoroutinefunction(f)
+
+
+ASYNC_MODE = is_async_mode()

aredis_om/util.py

@@ -5,7 +5,7 @@
 
 ADDITIONAL_REPLACEMENTS = {
     "aredis_om": "redis_om",
-    "aioredis": "redis",
+    "async_redis": "sync_redis",
     ":tests.": ":tests_sync.",
     "pytest_asyncio": "pytest",
     "py_test_mark_asyncio": "py_test_mark_sync",
@@ -26,11 +26,12 @@ def main():
         ),
     ]
     filepaths = []
-    for root, _, filenames in os.walk(
-            Path(__file__).absolute().parent
-    ):
+    for root, _, filenames in os.walk(Path(__file__).absolute().parent):
         for filename in filenames:
-            if filename.rpartition(".")[-1] in ("py", "pyi",):
+            if filename.rpartition(".")[-1] in (
+                "py",
+                "pyi",
+            ):
                 filepaths.append(os.path.join(root, filename))
 
     unasync.unasync_files(filepaths, rules)

make_sync.py

@@ -22,7 +22,7 @@ include=[
 ]
 
 [tool.poetry.dependencies]
-python = "^3.7"
+python = "^3.7,<=3.11"
 redis = ">=3.5.3,<5.0.0"
 aioredis = "^2.0.0"
 pydantic = "^1.8.2"
@@ -34,6 +34,7 @@ python-ulid = "^1.0.3"
 cleo = "1.0.0a5"
 typing-extensions = "^4.0.0"
 hiredis = "^2.0.0"
+more-itertools = "^8.14.0"
 
 [tool.poetry.dev-dependencies]
 mypy = "^0.971"

poetry.lock

@@ -16,14 +16,17 @@
     Field,
     HashModel,
     Migrator,
+    NotFoundError,
     QueryNotSupportedError,
     RedisModelError,
 )
 
 # We need to run this check as sync code (during tests) even in async mode
 # because we call it in the top-level module scope.
 from redis_om import has_redisearch
-from tests.conftest import py_test_mark_asyncio
+
+from .conftest import py_test_mark_asyncio
+
 
 if not has_redisearch():
     pytestmark = pytest.mark.skip
@@ -152,6 +155,23 @@ async def test_full_text_search_queries(members, m):
     assert actual == [member1, member3]
 
 
+@py_test_mark_asyncio
+async def test_pagination_queries(members, m):
+    member1, member2, member3 = members
+
+    actual = await m.Member.find(m.Member.last_name == "Brookins").page()
+
+    assert actual == [member1, member2]
+
+    actual = await m.Member.find().page(1, 1)
+
+    assert actual == [member2]
+
+    actual = await m.Member.find().page(0, 1)
+
+    assert actual == [member1]
+
+
 @py_test_mark_asyncio
 async def test_recursive_query_resolution(members, m):
     member1, member2, member3 = members
@@ -574,6 +594,35 @@ async def test_saves_many(m):
     assert await m.Member.get(pk=member2.pk) == member2
 
 
+@py_test_mark_asyncio
+async def test_delete_many(m):
+    member1 = m.Member(
+        id=0,
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        age=38,
+        bio="This is the user bio.",
+    )
+    member2 = m.Member(
+        id=1,
+        first_name="Kim",
+        last_name="Brookins",
+        email="k@example.com",
+        join_date=today,
+        age=34,
+        bio="This is the bio for Kim.",
+    )
+    members = [member1, member2]
+    result = await m.Member.add(members)
+    assert result == [member1, member2]
+    result = await m.Member.delete_many(members)
+    assert result == 2
+    with pytest.raises(NotFoundError):
+        await m.Member.get(pk=member1.pk)
+
+
 @py_test_mark_asyncio
 async def test_updates_a_model(members, m):
     member1, member2, member3 = members

pyproject.toml

@@ -25,7 +25,9 @@
 # We need to run this check as sync code (during tests) even in async mode
 # because we call it in the top-level module scope.
 from redis_om import has_redis_json
-from tests.conftest import py_test_mark_asyncio
+
+from .conftest import py_test_mark_asyncio
+
 
 if not has_redis_json():
     pytestmark = pytest.mark.skip
@@ -297,6 +299,33 @@ async def test_saves_many_explicit_transaction(address, m):
         assert await m.Member.get(pk=member2.pk) == member2
 
 
+@py_test_mark_asyncio
+async def test_delete_many_implicit_pipeline(address, m):
+    member1 = m.Member(
+        first_name="Andrew",
+        last_name="Brookins",
+        email="a@example.com",
+        join_date=today,
+        address=address,
+        age=38,
+    )
+    member2 = m.Member(
+        first_name="Kim",
+        last_name="Brookins",
+        email="k@example.com",
+        join_date=today,
+        address=address,
+        age=34,
+    )
+    members = [member1, member2]
+    result = await m.Member.add(members)
+    assert result == [member1, member2]
+    result = await m.Member.delete_many(members)
+    assert result == 2
+    with pytest.raises(NotFoundError):
+        await m.Member.get(pk=member2.pk)
+
+
 async def save(members):
     for m in members:
         await m.save()

tests/test_hash_model.py

@@ -9,7 +9,8 @@
 from pydantic import ValidationError
 
 from aredis_om import HashModel, Migrator, NotFoundError, RedisModelError
-from tests.conftest import py_test_mark_asyncio
+
+from .conftest import py_test_mark_asyncio
 
 
 today = datetime.date.today()

tests/test_json_model.py

@@ -0,0 +1,10 @@
+from aredis_om import redis
+from aredis_om.util import ASYNC_MODE
+
+
+def test_redis_type():
+    import redis as sync_redis_module
+    import redis.asyncio as async_redis_module
+
+    mapping = {True: async_redis_module, False: sync_redis_module}
+    assert mapping[ASYNC_MODE] is redis