FastAPI集成Supabase报错：APIResponse无法用于await表达式怎么解决？

场景与代码 #

在基于FastAPI和Supabase的项目中，编写了如下JournalLawService服务类：

class JournalLawService:
    def __init__(self):
        self.client = create_supabase_client()

    async def get_journal_law_by_id(self, law_id: int) -> Optional[JournalLaw]:
        try:
            result = (
                await self.client.table("Law").select("*").eq("id", law_id).execute()
            )
            return result.get("data", [None])[0]
        except Exception as e:
            print(f"Error fetching journal law by ID {law_id}: {e}")
            return None

报错信息 #

运行时出现以下错误：

Error fetching journal law by ID 5000: object APIResponse[~_ReturnT] can't be used in 'await' expression

问题分析 #

这是异步逻辑与Supabase客户端类型不匹配导致的问题：你当前使用的是Supabase的同步客户端，但在异步函数中用await调用了它的execute()方法——同步版本的execute()是普通函数，并非可等待对象（asyncio支持的coroutine或future），因此会触发该报错。移除返回类型注解、直接返回原始数据无法解决核心问题，因为问题根源在于客户端类型和调用方式不匹配。

解决方法 #

方法一：改用Supabase异步客户端（推荐） #

Supabase提供了专门的异步客户端，适配FastAPI的异步生态，步骤如下：

1. 导入异步客户端创建方法：

   from supabase import create_async_supabase_client
   

2. 修改服务类的初始化逻辑，创建异步客户端：

   class JournalLawService:
       def __init__(self):
           # 替换为你的Supabase URL和密钥
           self.client = create_async_supabase_client(SUPABASE_URL, SUPABASE_KEY)
   

3. 原有的查询逻辑无需修改，await可以正常作用于异步客户端的execute()方法。

方法二：在异步函数中包装同步调用（不推荐） #

如果暂时无法切换异步客户端，可以用asyncio.to_thread将同步调用放到单独线程中执行，避免阻塞FastAPI的事件循环：

import asyncio

class JournalLawService:
    def __init__(self):
        self.client = create_supabase_client()

    async def get_journal_law_by_id(self, law_id: int) -> Optional[JournalLaw]:
        try:
            result = await asyncio.to_thread(
                lambda: self.client.table("Law").select("*").eq("id", law_id).execute()
            )
            return result.get("data", [None])[0]
        except Exception as e:
            print(f"Error fetching journal law by ID {law_id}: {e}")
            return None

注意：此方法仅作为临时方案，长期来看推荐使用异步客户端，以更好地适配FastAPI的异步架构。

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