Acessing property_fields
that queries the database asynchronously
#443
-
I have a from typing import Optional, List
import ormar as orm
import databases
import sqlalchemy
DATABASE_URL = 'sqlite:///db.sqlite'
database = databases.Database(DATABASE_URL)
metadata = sqlalchemy.MetaData()
class BaseMeta(orm.ModelMeta):
metadata = metadata
database = database
class Person(orm.Model):
class Meta(BaseMeta):
tablename = "persons"
id: int = orm.Integer(primary_key=True)
name: str = orm.String(max_length=32)
@orm.property_field
def items_owned(self):
return len(self.items)
@orm.property_field
async def rare_items_owned(self):
return await self.items.filter(is_rare=True).count()
class Item(orm.Model):
class Meta(BaseMeta):
tablename = "items"
id: int = orm.Integer(primary_key=True)
name: str = orm.String(max_length=32)
is_rare: bool = orm.Boolean(default=False)
owner: Optional[Person] = orm.ForeignKey(Person, nullable=True)
async def create_db():
engine = sqlalchemy.create_engine(DATABASE_URL)
await database.connect()
metadata.drop_all(engine)
metadata.create_all(engine)
return metadata
async def tests():
person = await Person.objects.create(name='Igor')
item_a = await Item.objects.create(name='ItemA', owner=person, is_rare=True)
item_b = await Item.objects.create(name='ItemB', owner=person, is_rare=True)
item_c = await Item.objects.create(name='ItemC', owner=person)
item_d = await Item.objects.create(name='ItemD', owner=person)
# This works
print("Items owned: ", person.items_owned)
# This works
print("Rare Items owned: ", await person.rare_items_owned)
# This do not works =/
print(person.json(indent=2))
async def disconnect_db():
await database.disconnect()
async def main():
await create_db()
await tests()
await disconnect_db()
if __name__ == "__main__":
import asyncio
asyncio.run(main()) I'm able to call this How can I handle this case? |
Beta Was this translation helpful? Give feedback.
Replies: 1 comment 1 reply
-
Unfortunately, both python properties and ormar property_fields are sync only, as they are used in dumping the data into the dictionary in pydantic which is only synchronous. So in your case you can filter already loaded data: @orm.property_field
def rare_items_owned(self):
# items have to be already loaded at this point
return len([x for x in self.items if x.is_rare]) Otherwise, you have to fetch the loop and wrap the coroutine in |
Beta Was this translation helpful? Give feedback.
Unfortunately, both python properties and ormar property_fields are sync only, as they are used in dumping the data into the dictionary in pydantic which is only synchronous.
So in your case you can filter already loaded data:
Otherwise, you have to fetch the loop and wrap the coroutine in
asyncio.run()
call. (that should work)