WIP on async - test failure due to closed event loop

Author: Andrew Brookins

Makefile

@@ -1,4 +1,4 @@
-NAME := redis_developer
+NAME := redis_om
 INSTALL_STAMP := .install.stamp
 POETRY := $(shell command -v poetry 2> /dev/null)
 

README.md

@@ -52,7 +52,7 @@ Check out this example:
 import datetime
 from typing import Optional
 
-from redis_developer.model import (
+from redis_om.model import (
     EmbeddedJsonModel,
     JsonModel,
     Field,
@@ -172,9 +172,9 @@ Don't want to run Redis yourself? RediSearch and RedisJSON are also available on
 
 We'd love your contributions!
 
-**Bug reports** are especially helpful at this stage of the project. [You can open a bug report on GitHub](https://github.com/redis-developer/redis-developer-python/issues/new).
+**Bug reports** are especially helpful at this stage of the project. [You can open a bug report on GitHub](https://github.com/redis-om/redis-om-python/issues/new).
 
-You can also **contribute documentation** -- or just let us know if something needs more detail. [Open an issue on GitHub](https://github.com/redis-developer/redis-developer-python/issues/new) to get started.
+You can also **contribute documentation** -- or just let us know if something needs more detail. [Open an issue on GitHub](https://github.com/redis-om/redis-om-python/issues/new) to get started.
 
 ## License
 
@@ -184,17 +184,17 @@ Redis OM is [MIT licensed][license-url].
 
 [version-svg]: https://img.shields.io/pypi/v/redis-om?style=flat-square
 [package-url]: https://pypi.org/project/redis-om/
-[ci-svg]: https://img.shields.io/github/workflow/status/redis-developer/redis-developer-python/python?style=flat-square
-[ci-url]: https://github.com/redis-developer/redis-developer-python/actions/workflows/build.yml
+[ci-svg]: https://img.shields.io/github/workflow/status/redis-om/redis-om-python/python?style=flat-square
+[ci-url]: https://github.com/redis-om/redis-om-python/actions/workflows/build.yml
 [license-image]: http://img.shields.io/badge/license-MIT-green.svg?style=flat-square
 [license-url]: LICENSE
 
 <!-- Links -->
 
-[redis-developer-website]: https://developer.redis.com
-[redis-om-js]: https://github.com/redis-developer/redis-om-js
-[redis-om-dotnet]: https://github.com/redis-developer/redis-om-dotnet
-[redis-om-spring]: https://github.com/redis-developer/redis-om-spring
+[redis-om-website]: https://developer.redis.com
+[redis-om-js]: https://github.com/redis-om/redis-om-js
+[redis-om-dotnet]: https://github.com/redis-om/redis-om-dotnet
+[redis-om-spring]: https://github.com/redis-om/redis-om-spring
 [redisearch-url]: https://oss.redis.com/redisearch/
 [redis-json-url]: https://oss.redis.com/redisjson/
 [pydantic-url]: https://github.com/samuelcolvin/pydantic

build.py

@@ -0,0 +1,10 @@
+import unasync
+
+
+def build(setup_kwargs):
+    setup_kwargs.update(
+        {"cmdclass": {'build_py': unasync.cmdclass_build_py(rules=[
+            unasync.Rule("/aredis_om/", "/redis_om/"),
+            unasync.Rule("/aredis_om/tests/", "/redis_om/tests/", additional_replacements={"aredis_om": "redis_om"}),
+        ])}}
+    )

poetry.lock

@@ -1,9 +1,10 @@
 [tool.poetry]
-name = "redis-developer"
+name = "redis-om"
 version = "0.1.0"
 description = "A high-level library containing useful Redis abstractions and tools, like an ORM and leaderboard."
 authors = ["Andrew Brookins <[email protected]>"]
 license = "MIT"
+build = "build.py"
 
 [tool.poetry.dependencies]
 python = "^3.8"
@@ -30,10 +31,11 @@ bandit = "^1.7.0"
 coverage = "^6.0.2"
 pytest-cov = "^3.0.0"
 pytest-xdist = "^2.4.0"
-
+unasync = "^0.5.0"
+pytest-asyncio = "^0.16.0"
 
 [tool.poetry.scripts]
-migrate = "redis_developer.orm.cli.migrate:migrate"
+migrate = "redis_om.orm.cli.migrate:migrate"
 
 [build-system]
 requires = ["poetry-core>=1.0.0"]

pyproject.toml

@@ -1,22 +1,28 @@
 import os
+from typing import Union
 
 import dotenv
+import aioredis
 import redis
-
+from redis_om.unasync_util import ASYNC_MODE
 
 dotenv.load_dotenv()
 
 URL = os.environ.get("REDIS_OM_URL", None)
+if ASYNC_MODE:
+    client = aioredis.Redis
+else:
+    client = redis.Redis
 
 
-def get_redis_connection(**kwargs) -> redis.Redis:
+def get_redis_connection(**kwargs) -> Union[aioredis.Redis, 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 redis.from_url(url, **kwargs)
+        return client.from_url(url, **kwargs)
 
     # Decode from UTF-8 by default
     if "decode_responses" not in kwargs:
         kwargs["decode_responses"] = True
-    return redis.Redis(**kwargs)
+    return client(**kwargs)

redis_developer/__init__.py renamed to redis_om/__init__.py

@@ -1,10 +1,10 @@
 import click
 
-from redis_developer.model.migrations.migrator import Migrator
+from redis_om.model.migrations.migrator import Migrator
 
 
 @click.command()
-@click.option("--module", default="redis_developer")
+@click.option("--module", default="redis_om")
 def migrate(module):
     migrator = Migrator(module)
 

redis_developer/connections.py renamed to redis_om/connections.py

@@ -2,15 +2,14 @@
 import logging
 from dataclasses import dataclass
 from enum import Enum
-from typing import Optional
+from typing import Optional, Union
 
-from redis import ResponseError
+from redis import ResponseError, Redis
+from aioredis import ResponseError as AResponseError, Redis as ARedis
 
-from redis_developer.connections import get_redis_connection
-from redis_developer.model.model import model_registry
+from redis_om.model.model import model_registry
 
 
-redis = get_redis_connection()
 log = logging.getLogger(__name__)
 
 
@@ -43,12 +42,12 @@ def schema_hash_key(index_name):
     return f"{index_name}:hash"
 
 
-def create_index(index_name, schema, current_hash):
+async def create_index(redis: Union[Redis, ARedis], index_name, schema, current_hash):
     try:
-        redis.execute_command(f"ft.info {index_name}")
-    except ResponseError:
-        redis.execute_command(f"ft.create {index_name} {schema}")
-        redis.set(schema_hash_key(index_name), current_hash)
+        await redis.execute_command(f"ft.info {index_name}")
+    except (ResponseError, AResponseError):
+        await redis.execute_command(f"ft.create {index_name} {schema}")
+        await redis.set(schema_hash_key(index_name), current_hash)
     else:
         log.info("Index already exists, skipping. Index hash: %s", index_name)
 
@@ -65,34 +64,38 @@ class IndexMigration:
     schema: str
     hash: str
     action: MigrationAction
+    redis: Union[Redis, ARedis]
     previous_hash: Optional[str] = None
 
-    def run(self):
+    async def run(self):
         if self.action is MigrationAction.CREATE:
-            self.create()
+            await self.create()
         elif self.action is MigrationAction.DROP:
-            self.drop()
+            await self.drop()
 
-    def create(self):
+    async def create(self):
         try:
-            return create_index(self.index_name, self.schema, self.hash)
+            await create_index(self.redis, self.index_name, self.schema, self.hash)
         except ResponseError:
             log.info("Index already exists: %s", self.index_name)
 
-    def drop(self):
+    async def drop(self):
         try:
-            redis.execute_command(f"FT.DROPINDEX {self.index_name}")
+            await self.redis.execute_command(f"FT.DROPINDEX {self.index_name}")
         except ResponseError:
             log.info("Index does not exist: %s", self.index_name)
 
 
 class Migrator:
-    def __init__(self, module=None):
-        # Try to load any modules found under the given path or module name.
-        if module:
-            import_submodules(module)
-
+    def __init__(self, redis: Union[Redis, ARedis], module=None):
+        self.module = module
         self.migrations = []
+        self.redis = redis
+
+    async def run(self):
+        # Try to load any modules found under the given path or module name.
+        if self.module:
+            import_submodules(self.module)
 
         for name, cls in model_registry.items():
             hash_key = schema_hash_key(cls.Meta.index_name)
@@ -104,20 +107,21 @@ def __init__(self, module=None):
             current_hash = hashlib.sha1(schema.encode("utf-8")).hexdigest()  # nosec
 
             try:
-                redis.execute_command("ft.info", cls.Meta.index_name)
-            except ResponseError:
+                await self.redis.execute_command("ft.info", cls.Meta.index_name)
+            except (ResponseError, AResponseError):
                 self.migrations.append(
                     IndexMigration(
                         name,
                         cls.Meta.index_name,
                         schema,
                         current_hash,
                         MigrationAction.CREATE,
+                        self.redis
                     )
                 )
                 continue
 
-            stored_hash = redis.get(hash_key)
+            stored_hash = self.redis.get(hash_key)
             schema_out_of_date = current_hash != stored_hash
 
             if schema_out_of_date:
@@ -129,7 +133,8 @@ def __init__(self, module=None):
                         schema,
                         current_hash,
                         MigrationAction.DROP,
-                        stored_hash,
+                        self.redis,
+                        stored_hash
                     )
                 )
                 self.migrations.append(
@@ -139,12 +144,12 @@ def __init__(self, module=None):
                         schema,
                         current_hash,
                         MigrationAction.CREATE,
-                        stored_hash,
+                        self.redis,
+                        stored_hash
                     )
                 )
 
-    def run(self):
         # TODO: Migration history
         # TODO: Dry run with output
         for migration in self.migrations:
-            migration.run()
+            await migration.run()