基于 Flask Mega 教程的 Tweet Scheduler：获取 flask-sqlalchemy 数据库字段的问题

Question

我正在修改按照 Miguel Grinberg 的 Flask Mega 教程创建的 Flask 应用程序，以便可以 post 推文。我导入了用于访问 twitter api 的 tweepy 并修改了数据库以保存推文的预定时间。我希望在当前时间与预定时间匹配时迭代 current_user 的 post 和 SQLAlchemy 数据库中的相应时间以及 post 。

model.py中的数据库模型修改如下：

class Post(db.Model):
id = db.Column(db.Integer, primary_key = True)
body = db.Column(db.String(140))
timestamp = db.Column(db.DateTime, index=True, default=datetime.utcnow)
socialnetwork = db.Column(db.String(40))
user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
#This is the stuff for scheduling, just date
hour = db.Column(db.Integer)
minute = db.Column(db.Integer)
day = db.Column(db.Integer)
month = db.Column(db.Integer)
year = db.Column(db.Integer)
ampm = db.Column(db.String(2))

作为测试，我想遍历当前用户的 post 并使用 tweepy 向他们发送推文：

@app.before_first_request
def activate_job():
    def run_job():
        posts = current_user.followed_posts().filter_by(socialnetwork ='Twitter')
        for post in posts:
            tweepy_api.update_status(message)

            time.sleep(30)
    thread = threading.Thread(target=run_job)
    thread.start()

但是，这返回了错误：

AttributeError: 'NoneType' object has no attribute 'followed_posts'

在终端上。这让我感到困惑，因为我在同一个文件中多次使用 current_user 来通过社交网络过滤 post。

如下例routes.py

@app.route('/<username>')
@login_required
def user(username):
    user = User.query.filter_by(username = username).first_or_404()
    socialnetwork = request.args.get("socialnetwork")

    if socialnetwork == 'Facebook':

    posts = current_user.followed_posts().filter_by(socialnetwork = 'Facebook')

    elif socialnetwork == 'Twitter':
        posts = current_user.followed_posts().filter_by(socialnetwork = 'Twitter')
    else:
        posts = current_user.followed_posts()


    return render_template('user.html', user = user, posts = posts, form = socialnetwork)

上面的代码没有错误并且工作完美。

如果有人能指出我做错了什么，我将不胜感激。

Answer 1

您可能运行遇到问题，因为您正试图在不同的线程上获取 current_user（有关详细信息，请参阅 Flask docs）。您在没有任何当前用户的不同上下文中调用 run_job()（因为没有活动请求）。

我会对其进行修改，以便您在主线程中获取当前用户的帖子（即在 activate_job() 中，然后将帖子列表传递给后台进程进行处理。

类似于：

def activate_job():
    posts = current_user.followed_posts().filter_by(socialnetwork ='Twitter')
    def run_job(posts):
        for post in posts:
            tweepy_api.update_status(message)

            time.sleep(30)
    thread = threading.Thread(target=run_job, args=[posts])
    thread.start()

还值得注意的是，您可能需要重新考虑您的整体方法。与其检查每个请求是否有任何预定的推文要发送，不如使用某种独立于 Web 进程运行的后台任务队列。这样，您就不会对每个请求进行冗余检查，并且您不依赖于用户在预定时间发出请求。

参见 The Flask Mega-Tutorial Part XXII: Background Jobs for more details, and look into Celery。

基于 Flask Mega 教程的 Tweet Scheduler：获取 flask-sqlalchemy 数据库字段的问题

Tweet Scheduler based on Flask Megatutorial: Issues with getting fields of flask-sqlalchemy databse

flask

sqlalchemy

flask-sqlalchemy

twitter-oauth

sqlalchemy-migrate