如何在aiohttp中从URL获取查询字符串参数？代码报错求助

首先，你的代码核心问题出在路由配置错误，导致request.rel_url.query根本拿不到你想要的查询参数，另外还有SQL注入风险和资源泄漏的问题，咱们一步步来修复：

1. 修正路由配置 #

你现在把查询参数（call_id和status）当成了URL路径的一部分，写了/{call_id}&{status}，但实际上查询参数是跟在?后面的，比如正确的请求URL应该是http://localhost:8080/call-answer?call_id=123&status=completed。

所以路由应该改成普通的路径，比如：

app.router.add_get('/call-answer', call_answer)

这样当客户端发送带查询参数的请求时，request.rel_url.query才能正确解析出call_id和status。

2. 安全获取查询参数 #

直接用params["call_id"]如果参数不存在会抛出KeyError，建议先检查参数是否存在，或者用get方法设置默认值：

async def call_answer(request):
    params = request.rel_url.query
    # 先检查必要参数是否存在
    call_id = params.get("call_id")
    status = params.get("status")
    if not call_id or not status:
        return web.Response(
            content_type='application/json',
            status=400,
            text=json.dumps({
                'status': 'error',
                'message': 'Missing required parameters: call_id and status are required'
            }))
    # 后续逻辑...

3. 修复SQL注入风险 #

你现在用字符串拼接SQL的方式非常危险，会导致SQL注入攻击，psycopg2支持参数化查询，应该把参数作为第二个参数传给cursor.execute()：

# 错误写法（有注入风险）
# sql = """ update calls set status = %s where id = %s; """ % (params["status"], params["call_id"])
# cursor.execute(sql)

# 正确写法（参数化查询）
sql = """UPDATE calls SET status = %s WHERE id = %s;"""
cursor.execute(sql, (status, call_id))  # 把参数放在元组里传进去

4. 避免数据库资源泄漏 #

每次请求创建新的数据库连接不是最优解，而且你没有关闭cursor和connection，容易导致资源泄漏。建议用with语句自动管理资源，或者使用连接池：

async def call_answer(request):
    params = request.rel_url.query
    call_id = params.get("call_id")
    status = params.get("status")
    if not call_id or not status:
        return web.Response(
            content_type='application/json',
            status=400,
            text=json.dumps({
                'status': 'error',
                'message': 'Missing required parameters: call_id and status are required'
            }))
    
    try:
        conn = psycopg2.connect(connect_str)
        with conn.cursor() as cursor:
            sql = """UPDATE calls SET status = %s WHERE id = %s;"""
            cursor.execute(sql, (status, call_id))
            conn.commit()
        return web.Response(
            content_type='application/json',
            text=json.dumps({
                'status': 'success'
            }))
    except Exception as e:
        # 如果出错，回滚事务
        if conn:
            conn.rollback()
        return web.Response(
            content_type='application/json',
            status=500,
            text=json.dumps({
                'status': 'error',
                'message': str(e)
            }))
    finally:
        # 确保连接关闭
        if conn:
            conn.close()

完整修复后的代码 #

把以上修改整合起来，完整代码如下：

import argparse
import logging
import json
import psycopg2
from aiohttp import web

async def call_answer(request):
    params = request.rel_url.query
    call_id = params.get("call_id")
    status = params.get("status")
    
    # 验证必填参数
    if not call_id or not status:
        return web.Response(
            content_type='application/json',
            status=400,
            text=json.dumps({
                'status': 'error',
                'message': 'Missing required parameters: call_id and status are mandatory'
            }))
    
    try:
        # 假设connect_str已经定义
        connect_str = "your_connection_string_here"
        conn = psycopg2.connect(connect_str)
        with conn.cursor() as cursor:
            # 参数化查询，避免SQL注入
            sql = """UPDATE calls SET status = %s WHERE id = %s;"""
            cursor.execute(sql, (status, call_id))
            conn.commit()
        
        return web.Response(
            content_type='application/json',
            text=json.dumps({
                'status': 'success'
            }))
    except Exception as e:
        if conn:
            conn.rollback()
        return web.Response(
            content_type='application/json',
            status=500,
            text=json.dumps({
                'status': 'error',
                'message': f'Database error: {str(e)}'
            }))
    finally:
        if conn:
            conn.close()

if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='WebRTC audio / video / data-channels demo')
    parser.add_argument('--port', type=int, default=8080, help='Port for HTTP server (default: 8080)')
    parser.add_argument('--verbose', '-v', action='count')
    args = parser.parse_args()
    
    if args.verbose:
        logging.basicConfig(level=logging.DEBUG)
    
    app = web.Application()
    # 修正路由配置
    app.router.add_get('/call-answer', call_answer)
    
    web.run_app(app, port=args.port)

现在你可以用类似这样的请求测试：

http://localhost:8080/call-answer?call_id=1&status=answered

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