I can't get all the addresses associated with the customer.
class Customer(Base):
__tablename__ = "customer"
id = Column(Integer, primary_key=True)
authorization_token = Column(String, nullable=True)
name = Column(String, nullable=True)
surname = Column(String, nullable=True)
email = Column(String, nullable=True)
telephone = Column(String, nullable=True)
addresses = relationship("Address", secondary="customer_address", back_populates="customer")
async def to_dict(self):
return {"id": self.id,
"authorization_token": self.authorization_token,
"name": self.name,
"surname": self.surname,
"email": self.email,
"telephone": self.telephone,
"addresses": [address.to_dict() for address in self.addresses] }
class CustomerAddress(Base):
__tablename__ = "customer_address"
customer_id = Column(Integer, ForeignKey("customer.id"), primary_key=True)
address_id = Column(Integer, ForeignKey("address.id"), primary_key=True)
class Address(Base):
__tablename__ = "address"
id = Column(Integer, primary_key=True, autoincrement=True)
city_id = Column(Integer, ForeignKey("city.id"), nullable=True)
street = Column(String, nullable=True)
house = Column(String, nullable=True)
floor = Column(Integer, nullable=True)
entrance = Column(Integer, nullable=True)
apartment = Column(Integer, nullable=True)
customer = relationship("Customer", secondary="customer_address", back_populates="addresses")
async def to_dict(self):
return {"id": self.id,
"city_id": self.city_id,
"street": self.street,
"house": self.house,
"floor": self.floor,
"entrance": self.entrance,
"apartment": self.apartment}
async def get_customer_by_id(customer_id: int, session: AsyncSession = async_db_session) -> dict:
customer = select(Customer).where(Customer.id == customer_id)
customer = await session.execute(customer)
customer = customer.scalars().first()
if not customer:
return {'error': 'No customer found with the specified ID'}
customer = await customerto_dict()
return customer.to_dict()
Error:
sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place? (Background on this error at: https://sqlalche.me/e/14/xd2s)
sys:1: RuntimeWarning: coroutine 'AsyncAdapt_asyncpg_cursor._prepare_and_execute' was never awaited
The error is related to this line:
"addresses": [address.to_dict() for address in self.addresses]
Is it possible to retrieve data using a single query (asynchronous version of SQLAlchemy)?
What I haven't tried.)
The code above, one of the 5 versions, but the error(s) is caused by wanting to get data from "two" linked tables (method to_dict()).
My goal is to get: {"id": id, "authorization_token": token, "name": name ... "addresses": [ {"id": id, "city_id": city}, {"id": id, "city_id": city} ]
