Python项目重构SQLite数据库的实战记录

def insert_device_data(device_id, status, timestamp):
    conn = sqlite3.connect('data.db')
    cursor = conn.cursor()
    # 注意看这里，字符串直接拼接！
    sql = f"insert into device_log (device_id, status, log_time) values ('{device_id}', '{status}', '{timestamp}')"
    cursor.execute(sql)
    conn.commit()
    conn.close()

.schema device_log

create table device_log (
    id integer primary key,
    device_id text,
    status text,
    log_time text,
    raw_data text
);

# models.py
from sqlalchemy import column, integer, string, json, datetime
from sqlalchemy.ext.declarative import declarative_base
import json

base = declarative_base()

class devicelog(base):
    __tablename__ = 'device_log'

    id = column(integer, primary_key=true)
    device_id = column(string(64), nullable=false, index=true)  # 加了索引，因为常按设备查询
    status = column(string(32), nullable=false)
    log_time = column(string(32))  # 暂时保持string，兼容旧数据
    raw_data = column(json)  # 使用json类型，sqlalchemy会自动处理序列化/反序列化

    # 添加一个属性，方便访问解析后的raw_data
    @property
    def data(self):
        # 如果raw_data已经是dict（用了json类型后），直接返回，否则尝试解析
        if isinstance(self.raw_data, dict):
            return self.raw_data
        try:
            return json.loads(self.raw_data) if self.raw_data else {}
        except json.jsondecodeerror:
            return {}

alembic revision --autogenerate -m "initial migration based on existing schema"

alembic upgrade head

class devicelog(base):
    __tablename__ = 'device_log'
    # ... 其他字段不变
    log_time = column(datetime)  # 改为datetime类型
    # device_id的索引已经在定义中，无需修改

alembic revision --autogenerate -m "add index on device_id and change log_time to datetime"

def upgrade():
    # ### commands auto generated by alembic - please adjust! ###
    op.create_index(op.f('ix_device_log_device_id'), 'device_log', ['device_id'], unique=false)
    # 注意这里！alembic检测到类型变化，生成了alter_column操作
    op.alter_column('device_log', 'log_time',
                   existing_type=sa.text(),
                   type_=sa.datetime(),
                   existing_nullable=true)
    # ### end alembic commands ###

import sqlalchemy as sa
from alembic import op
import sqlite3
from datetime import datetime

# revision identifiers, used by alembic.
revision = 'xxxxxx'
down_revision = 'yyyyyy'

def upgrade():
    # 1. 先添加索引，这个操作sqlite原生支持，是安全的
    op.create_index(op.f('ix_device_log_device_id'), 'device_log', ['device_id'], unique=false)

    # 2. 处理log_time列类型转换。我们需要自定义逻辑。
    conn = op.get_bind()

    # 首先，读取所有现有的id和log_time文本
    rows = conn.execute(sa.text("select id, log_time from device_log where log_time is not null")).fetchall()

    # 创建一个临时字典，存储解析后的时间。如果解析失败，则置为none
    updates = []
    for row_id, time_str in rows:
        dt = none
        # 尝试解析旧数据中可能的时间格式
        for fmt in ('%y-%m-%d %h:%m:%s', '%y/%m/%d %h:%m:%s', '%y-%m-%dt%h:%m:%s'):
            try:
                dt = datetime.strptime(time_str, fmt)
                break
            except valueerror:
                continue
        updates.append((row_id, dt))

    # sqlite的alter column不支持类型变更，alembic会模拟。我们让它执行。
    # 但为了数据正确，我们在alembic操作后，手动更新数据。
    # 注意：alembic的模拟操作会创建新表、复制数据，但复制时log_time列的数据还是文本。
    # 所以我们需要在alembic的alter_column操作之后，再执行我们的更新。

    # 实际上，更安全的做法是分步迁移，这里为了示例，我们采用一个策略：
    # 先让alembic执行它的模拟alter操作（结构变更），然后我们手动更新数据。
    # 但alembic的op.alter_column在sqlite环境下是一个大事务，我们不能在中间插入。
    # 因此，我选择另一种方法：禁用alembic的自动alter，完全手动处理。

    # 注释掉自动生成的alter_column行
    # op.alter_column('device_log', 'log_time',
    #                existing_type=sa.text(),
    #                type_=sa.datetime(),
    #                existing_nullable=true)

    # 手动执行sqlite的列变更流程（简化版，假设没有复杂约束）
    # 步骤略复杂，在实际项目中，我可能会选择保持text类型，在应用层进行解析。
    # 鉴于篇幅和示例的清晰性，我决定在本文中采取一个折中方案：
    # 保留log_time为string，但添加一个datetime的属性方法，并确保新数据按iso格式存储。
    # 这是一个重要的实战决策：当数据迁移风险过高时，在应用层兼容是更稳妥的选择。

    # 所以，我回滚了模型中对log_time的修改，仍然使用string。
    # 并在devicelog类中添加一个方法：
    # @property
    # def log_datetime(self):
    #     # ... 解析self.log_time字符串为datetime对象，提供none安全
    pass

def downgrade():
    op.drop_index(op.f('ix_device_log_device_id'), table_name='device_log')
    # 类型改回操作也相应移除

alembic upgrade head

from sqlalchemy import column, integer, string, json, datetime, create_engine, index
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import json
from datetime import datetime

base = declarative_base()

class devicelog(base):
    __tablename__ = 'device_log'

    id = column(integer, primary_key=true)
    device_id = column(string(64), nullable=false)
    status = column(string(32), nullable=false)
    log_time = column(string(32))  # 保持文本存储，兼容性第一
    raw_data = column(json, default=dict)  # 使用json类型，默认空字典

    # 定义索引，注意这里只是元数据，实际创建需要alembic迁移
    __table_args__ = (index('ix_device_log_device_id', 'device_id'), )

    @property
    def data(self):
        """获取解析后的raw_data"""
        if isinstance(self.raw_data, dict):
            return self.raw_data
        try:
            return json.loads(self.raw_data) if self.raw_data else {}
        except json.jsondecodeerror:
            return {}

    @property
    def log_datetime(self):
        """将log_time字符串转换为datetime对象，安全版本"""
        if not self.log_time:
            return none
        # 尝试常见格式
        for fmt in ('%y-%m-%d %h:%m:%s', '%y/%m/%d %h:%m:%s', '%y-%m-%dt%h:%m:%s', '%y-%m-%d %h:%m:%s.%f'):
            try:
                return datetime.strptime(self.log_time, fmt)
            except valueerror:
                continue
        # 如果都无法解析，记录警告或返回none
        # import logging
        # logging.warning(f"无法解析时间字符串: {self.log_time}")
        return none

    @log_datetime.setter
    def log_datetime(self, dt: datetime):
        """设置datetime，统一存储为iso格式字符串"""
        if dt is none:
            self.log_time = none
        else:
            # 存储为iso格式，便于解析和排序
            self.log_time = dt.isoformat(sep=' ', timespec='seconds')  # 例如 '2023-10-27 14:30:00'

# 数据库连接和会话工厂
engine = create_engine('sqlite:///data.db', echo=false)  # echo=true可查看sql日志
sessionlocal = sessionmaker(bind=engine)

def get_db():
    """依赖注入使用的会话获取函数"""
    db = sessionlocal()
    try:
        yield db
    finally:
        db.close()

from models import devicelog, get_db
from datetime import datetime

# 插入新数据
with next(get_db()) as db:
    new_log = devicelog(
        device_id="device_001",
        status="online",
        log_datetime=datetime.now(),  # 使用setter
        raw_data={"temperature": 23.5, "humidity": 60}  # 直接传字典
    )
    db.add(new_log)
    db.commit()
    print(f"插入记录id: {new_log.id}")

# 查询数据
with next(get_db()) as db:
    logs = db.query(devicelog).filter(devicelog.device_id == "device_001").order_by(devicelog.log_time).all()
    for log in logs:
        print(f"id: {log.id}, status: {log.status}")
        print(f"  时间对象: {log.log_datetime}")  # 使用property获取datetime
        print(f"  原始数据: {log.data}")  # 使用property获取解析后的dict
        print(f"  存储的时间文本: {log.log_time}")