CODEWITHSUNDEEP
sqlalchemy
aik3e
aik3e

Reputation: 131

How to define cte/query/join in hybrid_property in sqlalchemy without repeating for each row in the model?

Models

class Category(Base):
__tablename__ = 'category'
__table_args__ = ({'schema': 'management'},)
category_id_seq = Sequence('management.category_id_seq')
id = Column(Integer, category_id_seq, server_default=category_id_seq.next_value(),
            primary_key=True, unique=True, nullable=False, )
description = Column(String, nullable=False)


class Brand(Base):
    __tablename__ = 'brand'
    __table_args__ = ({'schema': 'management'},)
    brand_id_seq = Sequence('management.brand_id_seq')
    id = Column(Integer, brand_id_seq, server_default=brand_id_seq.next_value(),
                primary_key=True, unique=True, nullable=False)
    description = Column(String, nullable=False)


class Quality(Base):
    __tablename__ = 'quality'
    __table_args__ = ({'schema': 'management'},)
    quality_id_seq = Sequence('management.quality_id_seq')
    id = Column(Integer, quality_id_seq, server_default=quality_id_seq.next_value(),
                primary_key=True, unique=True, nullable=False)
    description = Column(String, nullable=False)

class Product(Base):
    # region
    __tablename__ = 'product'
    __table_args__ = (UniqueConstraint('category_id', 'brand_id', 'description',
                                       name='product_uc'), {'schema': 'operation'})
    product_id_seq = Sequence('operation.product_id_seq')
    id = Column(Integer, product_id_seq, server_default=product_id_seq.next_value(),
                primary_key=True, unique=True, nullable=False)
    category_id = Column(Integer, ForeignKey(
        'management.category.id'), nullable=False)
    brand_id = Column(Integer, ForeignKey(
        'management.brand.id'), nullable=False)
    description = Column(String, nullable=False)
    category = relationship(
        'Category', foreign_keys='[Product.category_id]')
    brand = relationship(
        'Brand', foreign_keys='[Product.brand_id]')
    # endregion

    @hybrid_property
    def full_desc(self):
        cte_product = session.query(func.concat(Category.description, ' - ',
                                                Brand.description, ' - ',
                                                self.description).label('desc')) \
                                    .select_from(self) \
                                    .join(Category, Brand) \
                                    .filter(Category.id == self.category_id, Brand.id == self.brand_id) \
                                    .cte('cte_product')
        # return session.query(cte_product.c.desc).select_from(self).join(cte_product).filter(self.id == cte_product.c.id).cte('test')
        return cte_product.c.desc

    @full_desc.expression
    def full_desc(cls):
        cte_product = session.query(func.concat(Category.description, ' - ',
                                                Brand.description, ' - ',
                                                cls.description).label('desc')) \
                                    .select_from(cls)\
                                    .join(Category, Brand) \
                                    .filter(Category.id == cls.category_id, Brand.id == cls.brand_id) \
                                    .cte('cte_product')
        return cte_product.c.desc

Test 1 - Using hybrid_property - Undesired Results

result_1 = session.query(Product.id, Product.full_desc).all()
for i in result_1:
    print(i)

Running the code above results in a list of all 'full_desc' repeated for every single 'id', like so:

(1, 'Category6 - Brand1 - Product1')
(2, 'Category6 - Brand1 - Product1')
(3, 'Category6 - Brand1 - Product1')
(4, 'Category6 - Brand1 - Product1')
(5, 'Category6 - Brand1 - Product1')
(6, 'Category6 - Brand1 - Product1')
(7, 'Category6 - Brand1 - Product1')
(8, 'Category6 - Brand1 - Product1')
(9, 'Category6 - Brand1 - Product1')
(1, 'Category5 - Brand2 - Product2')
(2, 'Category5 - Brand2 - Product2')
(3, 'Category5 - Brand2 - Product2')
(4, 'Category5 - Brand2 - Product2')
... 81 rows in total

CTE defined outside models

cte_product = session.query(Product.id.label('id'), func.concat(Category.description, ' - ',
                                        Brand.description, ' - ',
                                        Product.description).label('desc')) \
                            .select_from(Product) \
                            .join(Category, Brand) \
                            .filter(Category.id == Product.category_id, Brand.id == Product.brand_id) \
                            .cte('cte_product')

Test 2 - Using CTE definition above (outside of Models) - Desired Results - Undesired Method

result_2 = session.query(cte_product.c.id, cte_product.c.desc).all()
for i in result_2:
    print(i)

Running the code above results in a list of right desc for each id with no repetition (only 9 rows), like so:

(1, 'Category6 - Brand1 - Product1')
(2, 'Category5 - Brand2 - Product2')
(3, 'Category7 - Brand4 - Product3')
(4, 'Category7 - Brand3 - Product4')
(5, 'Category5 - Brand1 - Product5')
(6, 'Category7 - Brand5 - Product6')
(7, 'Category3 - Brand2 - Product7')
(8, 'Category1 - Brand3 - Product8')
(9, 'Category4 - Brand3 - Product9')

Question

What should I change on the @hybrid_property and/or @hybridproperty.extension to get to the desired result as in Test 2 but using the methods in Test 1? Or is there a better way of doing this altogether?

Environment

SQLAlchemy==1.3.20
SQLAlchemy-Utils==0.36.7
PostgreSQL 13

Upvotes: 0

Views: 226

Answers (1)

van
van

Reputation: 76992

I am not sure if you should use cte at all. And the non-expression part of the @hybrid_property should not use any queries.

Please see the code below, which should work (using sqlalchemy version 1.4):

@hybrid_property
def full_description(self):
    return self.category.description + " - " + self.brand.description + " - " + self.description

@full_description.expression
def full_description(cls):
    subq = (
        select(
            func.concat(
                Category.description,
                " - ",
                Brand.description,
                " - ",
                cls.description,
            )
        )
        .where(Category.id == cls.category_id).where(Brand.id == cls.brand_id)
        .scalar_subquery()
        .label("full_desc_subquery")
    )
    return subq

and respective queries:

# in-memory (not using expressions, but potentially loading categories and brands from the database)
result_1 = session.query(Product).all()
for product in result_1:
    print(product.id, product.full_description)

# using .expression - get 'full_description' in one query.    
result_2 = session.query(Product.id, Product.full_description).all()
for product_id, fulldesc in result_2:
    print(product_id, fulldesc)

, where the latter would produce the following SQL statement:

SELECT product.id,

  (SELECT concat(category.description, %(concat_2)s, brand.description, %(concat_3)s, product.description) AS concat_1
   FROM category,
        brand
   WHERE category.id = product.category_id
     AND brand.id = product.brand_id) AS full_desc_subquery

FROM product

Upvotes: 1

Related Questions