Python数据库学习心得：SQLite、MySQL、PostgreSQL优缺点

import sqlite3

# 连接到SQLite数据库
# 如果数据库不存在，会自动创建
conn = sqlite3.connect('example.db')

# 创建游标
cursor = conn.cursor()

# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    age INTEGER,
    email TEXT UNIQUE
)
''')

# 插入数据
cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", ('Alice', 25, 'alice@example.com'))
cursor.execute("INSERT INTO users (name, age, email) VALUES (?, ?, ?)", ('Bob', 30, 'bob@example.com'))

# 提交事务
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# 更新数据
cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, 'Alice'))
conn.commit()

# 删除数据
cursor.execute("DELETE FROM users WHERE name = ?", ('Bob',))
conn.commit()

# 关闭连接
conn.close()

# 安装pymysql
# pip install pymysql

import pymysql

# 连接到MySQL数据库
conn = pymysql.connect(
    host='localhost',
    user='root',
    password='password',
    database='test',
    charset='utf8mb4',
    cursorclass=pymysql.cursors.DictCursor
)

# 创建游标
cursor = conn.cursor()

# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id INT AUTO_INCREMENT PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    age INT,
    email VARCHAR(255) UNIQUE
)
''')

# 插入数据
cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ('Alice', 25, 'alice@example.com'))
cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ('Bob', 30, 'bob@example.com'))

# 提交事务
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# 更新数据
cursor.execute("UPDATE users SET age = %s WHERE name = %s", (26, 'Alice'))
conn.commit()

# 删除数据
cursor.execute("DELETE FROM users WHERE name = %s", ('Bob',))
conn.commit()

# 关闭连接
conn.close()

# 安装psycopg2
# pip install psycopg2-binary

import psycopg2

# 连接到PostgreSQL数据库
conn = psycopg2.connect(
    host='localhost',
    user='postgres',
    password='password',
    database='test',
    port=5432
)

# 创建游标
cursor = conn.cursor()

# 创建表
cursor.execute('''
CREATE TABLE IF NOT EXISTS users (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    age INTEGER,
    email VARCHAR(255) UNIQUE
)
''')

# 插入数据
cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ('Alice', 25, 'alice@example.com'))
cursor.execute("INSERT INTO users (name, age, email) VALUES (%s, %s, %s)", ('Bob', 30, 'bob@example.com'))

# 提交事务
conn.commit()

# 查询数据
cursor.execute("SELECT * FROM users")
rows = cursor.fetchall()
for row in rows:
    print(row)

# 更新数据
cursor.execute("UPDATE users SET age = %s WHERE name = %s", (26, 'Alice'))
conn.commit()

# 删除数据
cursor.execute("DELETE FROM users WHERE name = %s", ('Bob',))
conn.commit()

# 关闭连接
conn.close()

# 安装SQLAlchemy
# pip install SQLAlchemy

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建数据库引擎
# SQLite
engine = create_engine('sqlite:///example.db')
# MySQL
# engine = create_engine('mysql+pymysql://root:password@localhost/test')
# PostgreSQL
# engine = create_engine('postgresql://postgres:password@localhost/test')

# 创建基类
Base = declarative_base()

# 定义模型
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    age = Column(Integer)
    email = Column(String(255), unique=True)

# 创建表
Base.metadata.create_all(engine)

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

# 插入数据
user1 = User(name='Alice', age=25, email='alice@example.com')
user2 = User(name='Bob', age=30, email='bob@example.com')
session.add(user1)
session.add(user2)
session.commit()

# 查询数据
users = session.query(User).all()
for user in users:
    print(user.id, user.name, user.age, user.email)

# 更新数据
user = session.query(User).filter_by(name='Alice').first()
user.age = 26
session.commit()

# 删除数据
user = session.query(User).filter_by(name='Bob').first()
session.delete(user)
session.commit()

# 关闭会话
session.close()

# models.py
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=255)
    age = models.IntegerField()
    email = models.EmailField(unique=True)

    def __str__(self):
        return self.name

# 使用Django ORM
# 创建用户
user = User.objects.create(name='Alice', age=25, email='alice@example.com')

# 查询用户
users = User.objects.all()
for user in users:
    print(user)

# 更新用户
user = User.objects.get(name='Alice')
user.age = 26
user.save()

# 删除用户
user = User.objects.get(name='Alice')
user.delete()

# 安装SQLAlchemy
# pip install SQLAlchemy

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

# 创建带有连接池的引擎
engine = create_engine(
    'mysql+pymysql://root:password@localhost/test',
    poolclass=QueuePool,
    pool_size=10,
    max_overflow=20,
    pool_pre_ping=True
)

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

# 使用会话
# ...

# 关闭会话
session.close()