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.
New contributor
João is a new contributor to this site. Take care in asking for clarification, commenting, and answering.
Check out our Code of Conduct.