Блог

Я разрабатываю API, который возвращает JSON, используя Flask, Sqlalchemy и базу данных MySQL, используемую внешним приложением. Проблема в том, что при новом запросе данных к API, который я создаю, данные, возвращаемые с использованием следующего кода, являются старыми (новые записи, добавленные внешним клиентом после запуска api, не отображаются).:

area/routes.py:

 from app.db import Base, session
from app.db.models import Room, Entry
@areas_bp.route('/area/<int:area_id>')
def area(area_id):
    '''
    Return all rooms in area and their entries
    :param area_id: id of an area
    :return:
    '''

    rooms = session.query(Room).filter_by(lsb_area_id=area_id)
    entries = session.query(Entry)

    session.rollback()

    ret_list = list()

    for room in rooms:
        entries_filtered = entries.filter_by(room_id=room.id)

        # ret_str  = f"Room {room.id} '{room.room_name}' <br />"
        room_dict = room.to_dict()
        room_dict['entries'] = [simplify_area_dict(entry.to_dict()) for entry in entries_filtered]

        ret_list.append(room_dict)

    return jsonify(ret_list)
 

Как вы можете видеть, я добавил session.rollback (), который, похоже, обновляет данные в сеансе, но я не уверен, что это эффективный способ достичь этого.
Я также читал о scoped_session, но я не уверен, поможет ли это мне каким-либо образом.
Можете ли вы указать мне правильное направление?

Спасибо!

настройка базы данных (app.db):

 from configparser import ConfigParser
from os.path import abspath

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, scoped_session


def read_db_config(filename='app/db/config.ini', section='mysql'):
    """ Read database configuration file and return a dictionary object
    :param filename: name of the configuration file
    :param section: section of database configuration
    :return: a dictionary of database parameters
    """
    # create parser and read ini configuration file
    parser = ConfigParser()
    parser.read(filename)

    # get section, default to mysql
    db = {}
    if parser.has_section(section):
        items = parser.items(section)
        for item in items:
            db[item[0]] = item[1]
    else:
        raise Exception('[MySQL] {0} not found in the {1} file. Abs filename: {2}'.format(section, filename, abspath(filename)))

    return db


db_config = read_db_config()

# SqlAlchemy Database Configuration With Mysql
db_uri = f"mysql pymysql://{ db_config['user'] }:{ db_config['password'] }@" 
                                        f"{ db_config['host'] }:{ db_config['port'] }/" 
                                        f"{ db_config['database'] }"
engine = create_engine(db_uri)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
session = scoped_session(SessionLocal)

Base = declarative_base()
metadata = Base.metadata