Sqlalchemy usage related

Paging during sub query

sub_query = db.session.query(Table,Table.id).offset((page_no - 1) * page_size).limit(page_size).subquery()
query1 = db.session.query(sub_query,  Table4, Table3, Table2) \
                .outerjoin(Table2) \
                .outerjoin(Table3) \
                .outerjoin(Table4) \

SQLAlchemy Inquiry exists

SELECT *
FROM table1 
WHERE EXISTS (SELECT * 
FROM (SELECT max(table2.event_type) AS event_type, table2.event_code AS event_code 
FROM table2 
WHERE table2.`status` = 200 GROUP BY table2.event_code) AS anon_1 
WHERE table1.id = anon_1.event_code AND anon_1.event_type = 100  ) ORDER BY table1.create_time DESC

query = db.session.query(Table1).order_by(Table1.create_time.desc())

func_criteria = db.session.query(func.max(Table2.event_type).label('event_type'),
        	Table2.event_code.label('event_code'))\
 			.group_by(Table2.event_code).filter(Table2.status == 200).subquery()
     
 query = query.filter(exists().where(Table1.id == func_criteria.c.event_code)\
 		.where(func_criteria.c.event_type== status))

Example of official website

>>> subq = (
...     select(func.count(address_table.c.id)).
...     where(user_table.c.id == address_table.c.user_id).
...     group_by(address_table.c.user_id).
...     having(func.count(address_table.c.id) > 1)
... ).exists()
>>> with engine.connect() as conn:
...     result = conn.execute(
...         select(user_table.c.name).where(subq)
...     )
...     print(result.all())

sqlalchemy

sqlalchemy Of update Manual commit Passed before yield When performing a remote asynchronous call session Object is replaced

SQLAlchemy==1.2.18

result = db.session.query(ObjectModel).get(ids.get("id"))
result.status = 2   # here  session._is_clean()  =  False

# Execute once in the middle yield The remote invocation 
# External request function 
def onAyncFetch():
    req = ... Omit other code 
    response = yield async_http.fetch(req)
    ... Omit other code 
    return Return(json_decode(response.body))
print id(result.query.session) 
data = yield onAyncFetch(...)
print id(result.query.session) # The memory address of the two prints is different 
... Omit other code 
# perform yield At the end of the call  session._is_clean()  =  True

db.session.commit() # Proceed again commit The following code segment will be carried out , It's modified above status Field cannot be commit

It's a piece of pseudo code . The process is roughly like this , Next paste sqlalchemy when commiit Source code

sqlalchemy/orm/session.py

    def _prepare_impl(self):
        self._assert_active()
        if self._parent is None or self.nested:
            self.session.dispatch.before_commit(self.session)

        stx = self.session.transaction
        if stx is not self:
            for subtransaction in stx._iterate_self_and_parents(upto=self):
                subtransaction.commit()

        if not self.session._flushing:
            for _flush_guard in range(100):
                if self.session._is_clean():  #  Judge whether the object information is changed , If there is a change, execute flush()
                    break
                self.session.flush()
            else:
                raise exc.FlushError(
                    "Over 100 subsequent flushes have occurred within "
                    "session.commit() - is an after_flush() hook "
                    "creating new objects?"
                )

        if self._parent is None and self.session.twophase:
            try:
                for t in set(self._connections.values()):
                    t[1].prepare()
            except:
                with util.safe_reraise():
                    self.rollback()

        self._state = PREPARED