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)  # 加了索引，因為常按設(shè)備查詢
    status = Column(String(32), nullable=False)
    log_time = Column(String(32))  # 暫時保持String，兼容舊數(shù)據(jù)
    raw_data = Column(JSON)  # 使用JSON類型，SQLAlchemy會自動處理序列化/反序列化

    # 添加一個屬性，方便訪問解析后的raw_data
    @property
    def data(self):
        # 如果raw_data已經(jīng)是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的索引已經(jīng)在定義中，無需修改

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列類型轉(zhuǎn)換。我們需要自定義邏輯。
    conn = op.get_bind()

    # 首先，讀取所有現(xiàn)有的id和log_time文本
    rows = conn.execute(sa.text("SELECT id, log_time FROM device_log WHERE log_time IS NOT NULL")).fetchall()

    # 創(chuàng)建一個臨時字典，存儲解析后的時間。如果解析失敗，則置為None
    updates = []
    for row_id, time_str in rows:
        dt = None
        # 嘗試解析舊數(shù)據(jù)中可能的時間格式
        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會模擬。我們讓它執(zhí)行。
    # 但為了數(shù)據(jù)正確，我們在Alembic操作后，手動更新數(shù)據(jù)。
    # 注意：Alembic的模擬操作會創(chuàng)建新表、復(fù)制數(shù)據(jù)，但復(fù)制時log_time列的數(shù)據(jù)還是文本。
    # 所以我們需要在Alembic的alter_column操作之后，再執(zhí)行我們的更新。

    # 實際上，更安全的做法是分步遷移，這里為了示例，我們采用一個策略：
    # 先讓Alembic執(zhí)行它的模擬alter操作（結(jié)構(gòu)變更），然后我們手動更新數(shù)據(jù)。
    # 但Alembic的op.alter_column在SQLite環(huán)境下是一個大事務(wù)，我們不能在中間插入。
    # 因此，我選擇另一種方法：禁用Alembic的自動alter，完全手動處理。

    # 注釋掉自動生成的alter_column行
    # op.alter_column('device_log', 'log_time',
    #                existing_type=sa.TEXT(),
    #                type_=sa.DateTime(),
    #                existing_nullable=True)

    # 手動執(zhí)行SQLite的列變更流程（簡化版，假設(shè)沒有復(fù)雜約束）
    # 步驟略復(fù)雜，在實際項目中，我可能會選擇保持TEXT類型，在應(yīng)用層進(jìn)行解析。
    # 鑒于篇幅和示例的清晰性，我決定在本文中采取一個折中方案：
    # 保留log_time為String，但添加一個DateTime的屬性方法，并確保新數(shù)據(jù)按ISO格式存儲。
    # 這是一個重要的實戰(zhàn)決策：當(dāng)數(shù)據(jù)遷移風(fēng)險過高時，在應(yīng)用層兼容是更穩(wěn)妥的選擇。

    # 所以，我回滾了模型中對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')
    # 類型改回操作也相應(yīng)移除

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類型，默認(rèn)空字典

    # 定義索引，注意這里只是元數(shù)據(jù)，實際創(chuàng)建需要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字符串轉(zhuǎn)換為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):
        """設(shè)置datetime，統(tǒng)一存儲為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'

# 數(shù)據(jù)庫連接和會話工廠
engine = create_engine('sqlite:///data.db', echo=False)  # echo=True可查看SQL日志
SessionLocal = sessionmaker(bind=engine)

def get_db():
    """依賴注入使用的會話獲取函數(shù)"""
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

from models import DeviceLog, get_db
from datetime import datetime

# 插入新數(shù)據(jù)
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}")

# 查詢數(shù)據(jù)
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"  原始數(shù)據(jù): {log.data}")  # 使用property獲取解析后的dict
        print(f"  存儲的時間文本: {log.log_time}")