FastAPI调用推文积分系统返回null 数据库操作无结果问题咨询

• summer.sum_fun() 返回的是整数类型的 total 值，不是类实例对象，接口中尝试读取 points.total 属于读取不存在的属性，直接导致返回 null
• main() 函数仅内部定义了 connect 和 append_to_database 两个工具函数，没有实际的调用逻辑，执行后不会拉取推特数据写入 Metrics 表，表为空导致积分核算无结果
• append_to_database 函数末尾递归调用自身，就算触发调用也会无限请求推特API，卡死流程无法走到积分返回步骤
• clear() 函数的数据库密码为 Mysql@123，和其余代码中的 pasword@123 不一致，大概率执行失败，无法正常清空表数据
• 无效导入 create_urls，原代码中对应函数名为 create_url，虽未触发调用但存在报错隐患

1. 修复积分返回逻辑 #

修改 summer.sum_fun() 方法，返回包含所有积分字段的字典，避免属性读取错误：

def sum_fun():
    try:
        con = mysql.connector.connect(host='localhost',
        database='twitterdb', user='root', password='pasword@123', charset='utf8')
        if con.is_connected():
            cursor = con.cursor(buffered=True)
            def sum_rts():
                cursor.execute("SELECT SUM(retweet_count) FROM twitterdb.Metrics")
                sum1=cursor.fetchall()[0][0]
                return int(sum1) if sum1 is not None else 0
            def sum_replies():
                cursor.execute("SELECT SUM(reply_count) FROM twitterdb.Metrics")
                sum2=cursor.fetchall()[0][0]
                return int(sum2) if sum2 is not None else 0
            def sum_likes():
                cursor.execute("SELECT SUM(like_count) FROM twitterdb.Metrics")
                sum3=cursor.fetchall()[0][0]
                return int(sum3) if sum3 is not None else 0
            def sum_qts():
                cursor.execute("SELECT SUM(quote_count) FROM twitterdb.Metrics")
                sum4=cursor.fetchall()[0][0]
                return int(sum4) if sum4 is not None else 0      
            like_points= 20 * sum_likes()
            reply_points= 100 * sum_replies()
            total_rts= sum_rts() + sum_qts()
            rt_points = 300 * total_rts
            total = like_points + reply_points + rt_points
            cursor.close()
            con.close()
            return {
                "like_points": like_points,
                "reply_points": reply_points,
                "rt_points": rt_points,
                "total": total
            }
    except Error as e:
        print(e)
        return {
            "like_points": 0,
            "reply_points": 0,
            "rt_points": 0,
            "total": 0
        }

2. 修复main函数执行逻辑 #

删除递归代码，补充实际调用逻辑：

def main():
    def connect(tweetid, retweet_count, reply_count, like_count, quote_count):
        # 原有插入逻辑不变，补全异常分支的连接处理
        try:
            con = mysql.connector.connect(host='localhost',
            database='twitterdb', user='root', password='pasword@123', charset='utf8')
            if con.is_connected():
                cursor = con.cursor(buffered=True)
                query = "INSERT INTO Metrics (tweetid,retweet_count,reply_count,like_count,quote_count) VALUES (%s, %s, %s, %s, %s)"
                cursor.execute(query, (tweetid,retweet_count,reply_count,like_count,quote_count))
                con.commit()
                cursor.close()
                con.close()
        except Error as e:
            print(e)
        return

    def append_to_database(json_response):
        for tweet in json_response['data']:  
            tweetid = tweet['id']
            retweet_count = tweet['public_metrics']['retweet_count']
            reply_count = tweet['public_metrics']['reply_count']
            like_count = tweet['public_metrics']['like_count']
            quote_count = tweet['public_metrics']['quote_count']
            connect(tweetid, retweet_count, reply_count, like_count, quote_count)
        # 删除以下三行递归代码
        # url = create_url()
        # json_response = connect_to_endpoint(url)
        # append_to_database(json_response)
    
    # 新增调用逻辑，避免ID为空时请求API报错
    if ids:
        url = create_url()
        json_response = connect_to_endpoint(url)
        append_to_database(json_response)

3. 修复clear函数 #

统一密码，补全事务提交和连接关闭逻辑：

def clear():
    try:
        con = mysql.connector.connect(host='localhost',
        database='twitterdb', user='root', password='pasword@123', charset='utf8')
        if con.is_connected():
            cursor = con.cursor(buffered=True)
            cursor.execute("truncate table twitterdb.Metrics")
            con.commit()
            cursor.close()
            con.close()
    except Error as e:
        print(e)
    return

4. 修复FastAPI接口逻辑 #

修改返回逻辑，清理无效导入，避免命名冲突：

from fastapi import FastAPI
from pydantic import BaseModel
from metrics import tweet_id, summer
import metrics
import uvicorn

app = FastAPI()

class UserReq(BaseModel):
    username:str       
     
@app.post('/calculate_points')
def calculate_points(user: UserReq):
    # 重置全局ID列表，避免多请求数据串扰
    metrics.ids = []
    tweets_list = tweet_id(user.username)
    tweets_list.get_tweet_ids(user.username)
    metrics.clear()
    metrics.main()
    points = summer.sum_fun()
    return points

if __name__ == "__main__":
    uvicorn.run("api:app", host="127.0.0.1", port=5000, log_level="info")

• 避免使用全局变量存储推文ID，可改为类属性或函数传参，防止多请求并发时数据串扰
• 复用数据库连接，无需每次操作都新建连接，提升性能
• 推特API单次最多支持查询100条推文的指标，超过100条需要做分页处理
• 敏感信息（数据库密码、Bearer Token）不要硬编码在代码中，建议通过环境变量读取
• 统一异常捕获和处理逻辑，避免接口无响应或返回不符合预期的结构

内容的提问来源于stack exchange，提问作者Niusoski