概述

supabase_config.py 文件提供了 Supabase 客户端和数据库连接池的配置与管理功能。

主要功能

• Supabase 客户端的创建与管理
• 异步数据库连接池的创建与管理
• 环境变量驱动的配置管理
• 数据库连接的安全处理

环境变量配置

数据库配置模块支持通过环境变量进行配置：

• SUPABASE_URL: Supabase 项目 URL
• SUPABASE_ANON_KEY: Supabase 匿名访问密钥
• DATABASE_URL: 数据库连接 URL
• DB_POOL_MIN_SIZE: 数据库连接池最小连接数，默认为 5
• DB_POOL_MAX_SIZE: 数据库连接池最大连接数，默认为 20
• DB_POOL_TIMEOUT: 数据库连接池超时时间，默认为 30 秒

使用方法

获取 Supabase 客户端

from app.utils.supabase_config import get_supabase_client

# 获取 Supabase 客户端实例
supabase = get_supabase_client()

# 使用客户端进行数据库操作
response = supabase.table('users').select('*').execute()

获取数据库连接池

import asyncio
from app.utils.supabase_config import get_db_pool

async def database_operation():
    # 获取数据库连接池
    pool = await get_db_pool()
    
    # 从连接池获取连接并执行操作
    async with pool.acquire() as connection:
        result = await connection.fetch("SELECT * FROM users")

# 运行异步函数
asyncio.run(database_operation())

关闭数据库连接池

import asyncio
from app.utils.supabase_config import close_db_pool

async def shutdown():
    # 应用关闭时关闭数据库连接池
    await close_db_pool()

# 运行关闭函数
asyncio.run(shutdown())

函数说明

create_db_pool()

创建异步数据库连接池。

返回值

• asyncpg.Pool: 数据库连接池实例

异常

• ValueError: 当 DATABASE_URL 环境变量未设置时抛出

get_db_pool()

获取数据库连接池实例，如果连接池未创建则先创建。

返回值

• asyncpg.Pool: 数据库连接池实例

close_db_pool()

关闭数据库连接池，释放所有连接资源。

get_supabase_client()

创建并返回 Supabase 客户端实例。

返回值

• Client: Supabase 客户端实例

示例

import os
import asyncio
from app.utils.supabase_config import (
    get_supabase_client, 
    get_db_pool, 
    close_db_pool
)

# 设置环境变量
os.environ['SUPABASE_URL'] = 'https://your-project.supabase.co'
os.environ['SUPABASE_ANON_KEY'] = 'your-anon-key'
os.environ['DATABASE_URL'] = 'postgresql://user:password@localhost:5432/database'

# 使用 Supabase 客户端
supabase = get_supabase_client()
users = supabase.table('users').select('*').execute()

# 使用数据库连接池
async def fetch_users():
    pool = await get_db_pool()
    async with pool.acquire() as connection:
        users = await connection.fetch("SELECT * FROM users")
        return users

users = asyncio.run(fetch_users())

# 应用关闭时清理资源
asyncio.run(close_db_pool())

版权所有

版权归属：Evoliant

许可证：MIT