Open belyak opened 1 year ago
Finally, in my project, I've used an overloaded db method, like this:
models.py:
import aredis_om
class MyBaseModel(aredis_om.JsonModel):
@classmethod
def db(cls):
return aredis_om.get_redis_connection()
class SampleModel(MyBaseModel):
field: str = aredis_om.Field(index=True)
class AnotherSampleModel(MyBaseModel):
field: int
You can add the Meta class to change the connection parameters without needed a BaseClass
from aredis_om import (
Field,
HashModel,
JsonModel,
EmbeddedJsonModel,
Migrator,
get_redis_connection,
)
redis_conn = get_redis_connection(
url=f"redis://10.9.9.4:6379", decode_responses=False, password="random"
)
class Matched(HashModel):
user_id1: str
user_id2: str
@root_validator()
def assign_pk(cls, values):
values["pk"] = f"{values['user_id1']}:{values['user_id2']}"
return values
class Meta:
database = redis_conn
Remember to call await Migrator().run()
It can be tricky to know when to call this to create indexes.
You can add the Meta class to change the connection parameters without needed a BaseClass
from aredis_om import ( Field, HashModel, JsonModel, EmbeddedJsonModel, Migrator, get_redis_connection, ) redis_conn = get_redis_connection( url=f"redis://10.9.9.4:6379", decode_responses=False, password="random" ) class Matched(HashModel): user_id1: str user_id2: str @root_validator() def assign_pk(cls, values): values["pk"] = f"{values['user_id1']}:{values['user_id2']}" return values class Meta: database = redis_conn
Remember to call
await Migrator().run()
It can be tricky to know when to call this to create indexes.
It will be validated only once then, the unit-testing assumes that all the resources are reinitialized for each case.
requirements.txt:
models.py:
func.py:
test_func.py:
outputs an error:
As you can see, the tests are the same, but the first is OK and the second fails. Besides of that, if to debug, there will be an exception "got Future attached to a different loop".
But if I change the test it works ok, however, in a real project it could be problematic to do like this:
models_hacked.py: