I'm having trouble trying to retrieve via query results from two related tables via a third mid table. It's a relationship between Product - Category via ProductCategory join I'm using SQLAlchemy, FastAPI and Alembic
The tables I'm using to create each entity in the Postgres
class ProductModel(Base):
__tablename__ = "products"
id = Column(String, primary_key=True, index=True, default=generate_uuid)
name = Column(String, index=True, nullable=False)
price = Column(Integer, nullable=False)
description = Column(String, nullable=False)
barcode = Column(String, unique=True, index=True, nullable=False)
section = Column(String, nullable=False)
stock = Column(Integer, nullable=False)
expire_date = Column(DateTime(timezone=True), nullable=False)
available = Column(Boolean, nullable=False)
images = relationship("ProductImages", back_populates="product")
categories = relationship("ProductCategoryJoin", back_populates="product")
orders = relationship("OrderProductJoin", back_populates="product")
class CategoryModel(Base):
__tablename__ = "categories"
id = Column(String, primary_key=True, index=True, default=generate_uuid)
name = Column(String, index=True, nullable=False)
products = relationship("ProductCategoryJoin", back_populates="category")
class ProductCategoryJoin(Base):
__tablename__ = "product_category_join"
product_id = Column(String, ForeignKey("products.id"), primary_key=True, nullable=False, autoincrement=False)
category_id = Column(String, ForeignKey("categories.id"), primary_key=True, nullable=False, autoincrement=False)
product = relationship("ProductModel", back_populates="categories")
category = relationship("CategoryModel", back_populates="products")
How I'm creating each of them in my service
def create_product(self, product: ProductBase):
try:
self.db.begin_nested()
product_instance = ProductCreate(
name=product.name,
price=product.price,
description=product.description,
barcode=product.barcode,
section=product.section,
stock=product.stock,
expire_date=product.expire_date,
available=product.available,
)
db_product = ProductModel(**product_instance.model_dump())
self.db.add(db_product)
self.db.flush()
for category in product.categories:
category_schema = self.get_category_by_id(category.id)
db_product_category = ProductCategoryJoin(product_id=str(db_product.id), category_id=category_schema.id)
self.db.add(db_product_category)
self.db.flush()
for image in product.images:
image = ProductImagesCreate(
image_url=image.image_url,
product_id=str(db_product.id)
)
db_image = ProductImages(**image.model_dump())
self.db.add(db_image)
self.db.flush()
self.db.commit()
self.db.refresh(db_product)
except IntegrityError:
self.db.rollback()
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Product already registered")
def create_category(self, category: CategoryBase):
category_data = CategoryCreate(name=category.name)
db_category = CategoryModel(\*\*category_data.model_dump())
try:
self.db.add(db_category)
self.db.commit()
self.db.refresh(db_category)
except IntegrityError:
self.db.rollback()
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Category already registered")
And this is the query I'm using to retrieve them
categories = (
self.db.query(CategoryModel).options(
joinedload(CategoryModel.products, innerjoin=True),
joinedload(ProductModel.images, innerjoin=True)).all()
)
I'm getting this error
sqlalchemy.exc.ArgumentError: Mapped class Mapper\[ProductModel(products)\] does not apply to any of the root entities in this query, e.g. Mapper\[CategoryModel(categories)\]. Please specify the full path from one of the root entities to the target attribute.
I can make it works like this, just for reference.
def get_categories(self):
categories = self.db.query(CategoryModel).options(joinedload(CategoryModel.products)).all()
categories_data = []
for category in categories:
products = category.products
product_list = []
for product_category_join in products:
product_list.append(Product(**product_category_join.product.__dict__)) # Access the ProductModel
categories_data.append(Category(name=category.name, id=category.id, products=product_list))
final_result = []
for category in categories_data:
final_result.append({
"id": category.id,
"name": category.name,
"products": [
{
"id": product.id,
"name": product.name,
"price": product.price,
"description": product.description,
"barcode": product.barcode,
"section": product.section,
"stock": product.stock,
"expire_date": product.expire_date.isoformat() if product.expire_date else None,
"available": product.available,
"images": [
{
'image_url': image.image_url
}
for image in self.db.query(ProductImages).filter(ProductImages.product_id == product.id).all()
]
}
for product in category.products
]
})
It isn't completely clear to me what you want to do here but I think you want to "chain" these loaders together. Ie. call one off the other and not list them separately with a comma. The error is saying that it doesn't know how to load from CategoryModel to ProductModel. If you chain the expressions and then add in the middle "association object" as well then the SQLAlchemy can figure out how to build the query.
I used an outerjoin for the last loader because otherwise I think some products wouldn't load if they didn't have an image.
categories = session.query(CategoryModel).options(
joinedload(
CategoryModel.products, innerjoin=True
).joinedload(
ProductCategoryJoin.product, innerjoin=True
).joinedload(
ProductModel.images)).all()
SELECT
categories.id AS categories_id,
categories.name AS categories_name,
product_images_1.id AS product_images_1_id,
product_images_1.product_id AS product_images_1_product_id,
products_1.id AS products_1_id,
products_1.name AS products_1_name,
product_category_join_1.product_id AS product_category_join_1_product_id,
product_category_join_1.category_id AS product_category_join_1_category_id
FROM
categories
JOIN
product_category_join AS product_category_join_1
ON
categories.id = product_category_join_1.category_id
JOIN
products AS products_1
ON
products_1.id = product_category_join_1.product_id
LEFT OUTER JOIN
product_images AS product_images_1
ON
products_1.id = product_images_1.product_id
Chaining is mentioned here near:
The loader options can also be “chained” using method chaining to specify how loading should occur further levels deep: