Python实现数据库连接池的代码示例和最佳实践

import psycopg2
from psycopg2 import pool

# 创建连接池
postgreSQL_pool = psycopg2.pool.SimpleConnectionPool(1, 20,
                                                     user="your_user",
                                                     password="your_password",
                                                     host="127.0.0.1",
                                                     port="5432",
                                                     database="your_database")

if (postgreSQL_pool):
    print("Connection pool created successfully")

# 获取连接
conn = postgreSQL_pool.getconn()
if (conn):
    print("successfully received connection from connection pool ")
    conn.autocommit = True
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM your_table")
    records = cursor.fetchall()
    print(records)
    # 归还连接
    postgreSQL_pool.putconn(conn)
    print("Put away a PostgreSQL connection")

# 关闭连接池
postgreSQL_pool.closeall()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建引擎
engine = create_engine('postgresql://your_user:your_password@localhost:5432/your_database', pool_size=10, max_overflow=20)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 执行查询
result = session.execute("SELECT * FROM your_table")
for row in result:
    print(row)

# 关闭会话
session.close()

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# 创建引擎
engine = create_engine('postgresql://your_user:your_password@localhost:5432/monkey_music', pool_size=20, max_overflow=10)

# 创建会话
Session = sessionmaker(bind=engine)
session = Session()

# 用户登录
def login(username, password):
    result = session.execute("SELECT * FROM users WHERE username = :username AND password = :password", {"username": username, "password": password})
    user = result.fetchone()
    if user:
        print(f"User {username} logged in successfully")
    else:
        print("Invalid credentials")

# 歌曲搜索
def search_songs(keyword):
    result = session.execute("SELECT * FROM songs WHERE title LIKE :keyword", {"keyword": f"%{keyword}%"})
    for song in result:
        print(song)

# 播放记录
def record_playback(user_id, song_id):
    session.execute("INSERT INTO playback_history (user_id, song_id) VALUES (:user_id, :song_id)", {"user_id": user_id, "song_id": song_id})
    session.commit()

# 示例调用
login("user1", "password1")
search_songs("love")
record_playback(1, 123)

# 关闭会话
session.close()