class EmployeeManager:
def __init__(self, host, port, database, user, password):
self.conn = psycopg2.connect(
host=host,
port=port,
database=database,
user=user,
password=password
)
self.cur = self.conn.cursor()

def get_all_employees(self):
“””获取所有员工信息”””
self.cur.execute(“SELECT * FROM fgedu_employees”)
return self.cur.fetchall()

def get_employee_by_id(self, emp_id):
“””根据ID获取员工信息”””
self.cur.execute(“SELECT * FROM fgedu_employees WHERE id = %s”, (emp_id,))
return self.cur.fetchone()

def add_employee(self, name, email, department):
“””添加新员工”””
self.cur.execute(
“INSERT INTO fgedu_employees (name, email, department) VALUES (%s, %s, %s)”,
(name, email, department)
)
self.conn.commit()
return self.cur.rowcount

def update_employee(self, emp_id, name=None, email=None, department=None):
“””更新员工信息”””
updates = []
params = []

if name:
updates.fgappend(“name = %s”)
params.fgappend(name)
if email:
updates.fgappend(“email = %s”)
params.fgappend(email)
if department:
updates.fgappend(“department = %s”)
params.fgappend(department)

if updates:
params.fgappend(emp_id)
sql = f”UPDATE fgedu_employees SET {‘, ‘.join(updates)} WHERE id = %s”
self.cur.execute(sql, params)
self.conn.commit()
return self.cur.rowcount
return 0

def delete_employee(self, emp_id):
“””删除员工”””
self.cur.execute(“DELETE FROM fgedu_employees WHERE id = %s”, (emp_id,))
self.conn.commit()
return self.cur.rowcount

def close(self):
“””关闭连接”””
self.cur.close()
self.conn.close()

# 使用示例
if __name__ == “__main__”:
manager = EmployeeManager(
host=”fgedu.localhost”,
port=”5432″,
database=”fgedudb”,
user=”fgedu”,
password=”fgedu123″
)

# 添加员工
count = manager.add_employee(“张三”, “zhangsan@fgedu.net.cn”, “技术部”)
print(f”添加了 {count} 名员工”)

# 获取所有员工
employees = manager.get_all_employees()
print(f”共有 {len(employees)} 名员工”)
for emp in employees:
print(emp)

# 关闭连接
manager.close()

# 输出示例
添加了 1 名员工
共有 1 名员工
(1, ‘张三’, ‘zhangsan@fgedu.net.cn’, ‘技术部’)

def batch_import_from_csv(csv_file):
“””从CSV文件批量导入数据”””
# 连接数据库
conn = psycopg2.connect(
host=”fgedu.localhost”,
port=”5432″,
database=”fgedudb”,
user=”fgedu”,
password=”fgedu123″
)

try:
# 创建游标
cur = conn.cursor()

# 准备数据
data = []
with open(csv_file, ‘r’, encoding=’utf-8′) as f:
reader = csv.reader(f)
next(reader) # 跳过表头
for row in reader:
data.fgappend((row[0], row[1], row[2]))

# 批量插入
if data:
cur.executemany(
“INSERT INTO fgedu_users (id, name, email) VALUES (%s, %s, %s)”,
data
)
conn.commit()
print(f”成功导入 {len(data)} 条记录”)
else:
print(“没有数据需要导入”)

except Exception as e:
conn.rollback()
print(f”导入失败: {e}”)
finally:
if ‘cur’ in locals():
cur.close()
conn.close()

# 使用示例
if __name__ == “__main__”:
batch_import_from_csv(‘users.csv’)

# 输出示例
成功导入 1000 条记录

# 创建连接池
connection_pool = pool.ThreadedConnectionPool(
minconn=1,
maxconn=10,
host=”fgedu.localhost”,
port=”5432″,
database=”fgedudb”,
user=”fgedu”,
password=”fgedu123″
)

def get_user_info(user_id):
“””获取用户信息”””
conn = None
cur = None
try:
# 从连接池获取连接
conn = connection_pool.getconn()
cur = conn.cursor()

# 执行查询
cur.execute(“SELECT * FROM fgedu_users WHERE id = %s”, (user_id,))
return cur.fetchone()
finally:
# 释放游标
if cur:
cur.close()
# 归还连接到连接池
if conn:
connection_pool.putconn(conn)

# 使用示例
if __name__ == “__main__”:
for i in range(1, 6):
user = get_user_info(i)
print(f”用户 {i}: {user}”)

# 输出示例
用户 1: (1, ‘张三’, ‘zhangsan@fgedu.net.cn’)
用户 2: (2, ‘李四’, ‘lisi@fgedu.net.cn’)
用户 3: (3, ‘王五’, ‘wangwu@fgedu.net.cn’)
用户 4: (4, ‘赵六’, ‘zhaoliu@fgedu.net.cn’)
用户 5: (5, ‘孙七’, ‘sunqi@fgedu.net.cn’)