将 SQLAlchemy 表导入没有 Flask 的 pandas 数据框

Question

我正在创建一个服务器应用程序，它将利用由 Flask GUI 创建的远程数据库。为了简单起见（因为它主要是线程计算），我想在不使用 Flask 框架的情况下构建我的服务器应用程序。我将有一个 redis 消息服务器来保持两个进程同步。

我的挑战是我想尽可能使用持久存储。因此，作为其中的一部分，我的服务器应用程序将需要导入 Flask-SQLAlchemy 数据库文件来检索参数。我遇到的困难是编写将 SQLAlchemy 文件拉入熊猫的导入语句。这是我目前所拥有的：

config.py

import os


class Config:
    SQLALCHEMY_DATABASE_URI_REMOTE = os.environ.get('RDQ_LOGIN') # remote docker database
    SQLALCHEMY_DATABASE_URI_LOCAL = os.environ.get('RDU_LOGIN') # local file database
    REDIS_IP = os.environ.get('REDIS_IP')
    REDIS_PASSWORD = os.environ.get('REDIS_PASSWORD')

初始化.py

from src_code.config import Config
import redis
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import orm
from sqlalchemy.ext.declarative import declarative_base
import sqlalchemy as sa


engine_remote = create_engine(Config.SQLALCHEMY_DATABASE_URI_REMOTE, echo=True)
Base_remote = declarative_base()
Base_remote.metadata.create_all(engine_remote)
Session_remote = sessionmaker(bind=engine_remote)
Session_remote.configure(bind=engine_remote)
session_remote = Session_remote()

engine_local = create_engine(Config.SQLALCHEMY_DATABASE_URI_LOCAL, echo=True)
Base_local = declarative_base()
Base_local.metadata.create_all(engine_local)
Session_local = sessionmaker(bind=engine_local)
Session_local.configure(bind=engine_local)
session_local = Session_local()

redisChannel = redis.StrictRedis(host=Config.REDIS_IP, port=6379, password=Config.REDIS_PASSWORD,
                                 decode_responses=True)

main.py

import pandas as pd
from src_code import session_local, session_remote

def start():
    table_df = pd.read_sql(
        'SELECT * from game',
        con=session_remote
    )
    print(table_df)

if __name__ == "__main__":
    start()

很遗憾，我遇到了一个错误：

AttributeError: 'Session' object has no attribute 'cursor'

我不需要能够写回表（所以只读就足够了）。所以我使用了一个简单的例子，它可以让我从 pandas 中提取我需要的东西（我对使用 pandas 比使用 SQL 更有信心）。

我确实有可以在服务器代码中复制的模型语句，如果这能以某种方式促进该过程：

from datetime import datetime
from itsdangerous import TimedJSONWebSignatureSerializer as Serializer
from flask import current_app
from src_code import db, login_manager
from flask_login import UserMixin


@login_manager.user_loader
def load_user(user_id):
    return User.query.get(int(user_id))


class User(db.Model, UserMixin):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(20), unique=True, nullable=False)
    email = db.Column(db.String(120), unique=True, nullable=False)
    agent = db.Column(db.Integer, unique=False, nullable=False)
    image_file = db.Column(db.String(20), unique=False, nullable=False,default='default.jpg')
    password = db.Column(db.String(60), nullable=False)
    posts = db.relationship('Post', backref='author', lazy=True)
    games_started = db.relationship('Game', backref='captained', lazy=True)
    games_played = db.relationship('Player', backref='games', lazy=True)

    def get_reset_token(self, expires_sec=1800):
        s = Serializer(current_app.config['SECRET_KEY'], expires_sec)
        return s.dumps({'user_id': self.id}).decode('utf-8')

    @staticmethod
    def verify_reset_token(token):
        s = Serializer(current_app.config['SECRET_KEY'])
        try:
            user_id = s.loads(token)['user_id']
        except:
            return None
        return User.query.get(user_id)

    def __repr__(self):
        return f"User('{self.username}', '{self.email}', '{self.agent}', '{self.image_file}')"


class Post(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(100), nullable=False)
    date_posted = db.Column(db.DateTime, nullable=False,default=datetime.utcnow)
    content = db.Column(db.Text, nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def __repr__(self):
        return f"Post('{self.title}', '{self.date_posted}')"


class Game(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    title = db.Column(db.String(60), nullable=False)
    date_posted = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    company_count = db.Column(db.Integer, nullable=False)
    starting_year = db.Column(db.Integer, nullable=False)
    time_limit = db.Column(db.Integer, nullable=False)
    agent_decisions_visible = db.Column(db.Boolean, nullable=False)
    client_count = db.Column(db.Integer, nullable=False)
    pre_game_yrs = db.Column(db.Integer, nullable=False)
    time_index = db.Column(db.Integer, nullable=False)
    game_active = db.Column(db.Boolean, nullable=False)
    player_capt = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    games_msgs = db.relationship('Messages', backref='messages', lazy=True)

    def __repr__(self):
        return f"Game('{self.title}', '{self.date_posted}')"


class GameRequests(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    date_requested = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def __repr__(self):
        return f"GameRequest('{self.game_id}', '{self.user_id}', '{self.date_requested}'"


class GameDecisions(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    date_requested = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)
    time_index = db.Column(db.Integer, nullable=False)


    def __repr__(self):
        return f"GameDecisions('{self.game_id}', '{self.user_id}', '{self.time_index}'"


class Player(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    date_posted = db.Column(db.DateTime, nullable=False, default=datetime.utcnow)
    game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False)

    def __repr__(self):
        return f"Player('{self.game_id}', '{self.user_id}', '{self.date_posted}'"


class Messages(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    date_posted = db.Column(db.DateTime, nullable=False,default=datetime.utcnow)
    type = db.Column(db.String(10), nullable=False)
    msg_from = db.Column(db.String(20), nullable=False)
    msg_to = db.Column(db.String(20), nullable=False)
    message = db.Column(db.String(60), nullable=False)
    game_id = db.Column(db.Integer, db.ForeignKey('game.id'), nullable=False)

    def __repr__(self):
        return f"Messages('{self.game_id}', '{self.id}', '{self.message}', '{self.date_posted}')"

感谢您提供的任何帮助！

Answer 1

与您的结论类似，以下是我将数据库读入 pandas 的方法：

# Create your query.
# This can be as complex or simple as you'd like
query = session_remote.query(Game)

df = pd.read_sql(query.statement, session_remote.bind)

这里的关键区别在于使用 ORM 来执行（或者更确切地说，编写）查询本身。

在 ORM 后面屏蔽 SQL 有 many advantages——我强烈建议不要在生产后端使用原始 SQL。

Answer 2

似乎我比我意识到的更接近。下面的代码做了我需要的：

SQLAlchemy ORM conversion to pandas DataFrame

table_df = pd.read_sql(
           'SELECT * from game',
           session_remote.bind
           )