tmtt/redis-om-python
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Migrate from aioredis to redis-py with asyncio support (#233)

Browse source 
* Migrate from aioredis to redis with asyncio support

Add test for redis type
Fix imports from wrong module (for tests_sync)

* fixing merge conflicts and up to dating the lock file

Co-authored-by: Chayim I. Kirshen <c@kirshen.com>
This commit is contained in:
Serhii Charykov 2022-08-10 15:21:13 +03:00 • committed by GitHub
parent e5e887229a
commit 4661459ddd
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
15 changed files with 484 additions and 494 deletions
Download patch file Download diff file Expand all files Collapse all files

39
aredis_om/model/migrations/migrator.py
View file

@ -4,7 +4,7 @@ from dataclasses import dataclass
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 @@ class IndexMigration:
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 @@ class Migrator:
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,8 +115,8 @@ class Migrator:
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,
@ -123,12 +124,12 @@ class Migrator:
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 @@ class Migrator:
schema,
current_hash,
MigrationAction.DROP,
redis,
conn,
stored_hash,
)
)
@ -151,7 +152,7 @@ class Migrator:
schema,
current_hash,
MigrationAction.CREATE,
redis,
conn,
stored_hash,
)
)

27
aredis_om/model/model.py
View file

@ -24,8 +24,6 @@ from typing import (
no_type_check,
)
import aioredis
from aioredis.client import Pipeline
from pydantic import BaseModel, validator
from pydantic.fields import FieldInfo as PydanticFieldInfo
from pydantic.fields import ModelField, Undefined, UndefinedType
@ -35,9 +33,10 @@ from pydantic.utils import Representation
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
@ -978,7 +977,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
@ -997,7 +996,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
@ -1130,10 +1129,14 @@ class RedisModel(BaseModel, abc.ABC, metaclass=ModelMeta):
"""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):
async def expire(
self, num_seconds: int, pipeline: Optional[redis.client.Pipeline] = None
):
if pipeline is None:
db = self.db()
else:
@ -1226,7 +1229,7 @@ class RedisModel(BaseModel, abc.ABC, metaclass=ModelMeta):
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:
@ -1286,7 +1289,9 @@ class HashModel(RedisModel, abc.ABC):
f"HashModels cannot index dataclass fields. Field: {name}"
)
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()
@ -1458,7 +1463,9 @@ class JsonModel(RedisModel, abc.ABC):
)
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()