SQLAlchemy Insert executemany Func

5 人关注

我正试图通过SQLAlchemy向MySQL数据库批量插入一个dict的列表。 在这个列表中,每个dict都有时间和温度。 时间值是一个ISO时间字符串。 我试图让MySQL使用STR_TO_DATE函数来创建数据时间对象以存储到数据库中。

我已经成功地通过insert().values({'time':func.STR_TO_DATE(row.time, date_str), 'temp' = row.temp})实现了这个工作。

示例代码。

import sqlalchemy
from sqlalchemy import create_engine, MetaData, Table, Column, func
from sqlalchemy.dialects.mysql import DATETIME, INTEGER, insert
temp_data = [{'time':'2019-02-07T14:01:00.000000000Z', 'temperature':68},
             {'time':'2019-02-07T14:02:00.000000000Z', 'temperature':69},
             {'time':'2019-02-07T14:03:00.000000000Z', 'temperature':70},
             {'time':'2019-02-07T14:04:00.000000000Z', 'temperature':71}]
date_string = '%Y-%m-%dT%H:%i:%S.%f000Z'
mysql_metadata = MetaData()
mysql_table = Table('temperature', mysql_metadata,
                    Column('time', DATETIME(fsp = 6), primary_key = True),
                    Column('temperature', INTEGER()))
def single_insert():
    engine = create_engine('mysql://test_user:test@localhost/temperature',
                                  echo = False)
    mysql_table.drop(engine, checkfirst = True)
    mysql_table.create(engine)
    conn = engine.connect()
    for row in temp_data:
        conn.execute(mysql_table.insert().values(time = func.STR_TO_DATE(row['time'], date_string),
                                                 temperature = row['temperature']))
def insert_many():
    engine = create_engine('mysql://test_user:test@localhost/temperature',
                                  echo = False)    
    mysql_table.drop(engine, checkfirst = True)
    mysql_table.create(engine)
    conn = engine.connect()
    conn.execute(mysql_table.insert(), temp_data)

不知道如何通过将一个参数绑定到一个可以调用STR_TO_DATE的单一输入值,使数据可以通过一个dict列表插入来实现。

Second Edit:

这是我的蛮力代码,可以工作。 试图找到一种方法来使用SQLALchemy核心来完成这个语句。

def insert_many():
    engine = create_engine('mysql://test_user:test@localhost/temperature',
                                  echo = True)    
    mysql_table.drop(engine, checkfirst = True)
    mysql_table.create(engine)
    conn = engine.connect()
    stmt = text("INSERT INTO temperature VALUES( STR_TO_DATE(:time, '%Y-%m-%dT%H:%i:%S.%f000Z'), :temperature)")
    conn.execute(stmt, temp_data)

Third Edit:

最终想对Python Datetime strptime与MySQL方法做一个剖析。 想看看哪个更快。

1 个评论
没有实际的代码,不知道问题出在哪里。请给出一个 最低限度的可重复的例子
python
mysql
sqlalchemy
Brian M.
Brian M.
发布于 2019-06-24
2 个回答
Robert Thapa
Robert Thapa
发布于 2020-01-19
已采纳
0 人赞同

对字典列表进行批量插入的一种方法是使用 bindparam 为每一个键,并执行执行

from sqlalchemy.sql.expression import bindparam
temp_data = [{'time':'2019-02-07T14:01:00.000000000Z', 'temperature':68},
         {'time':'2019-02-07T14:02:00.000000000Z', 'temperature':69},
         {'time':'2019-02-07T14:03:00.000000000Z', 'temperature':70},
         {'time':'2019-02-07T14:04:00.000000000Z', 'temperature':71}]
def insert_many():
    engine = create_engine('mysql://test_user:test@localhost/temperature',
                              echo = False)    
    mysql_table.drop(engine, checkfirst = True)
    mysql_table.create(engine)
    conn = engine.connect()
    statement = Mysql_table.insert().values({
            'time' : bindparam('time'),
            'temperature' : bindparam('temperature')
    conn.execute(statement, temp_data)
    conn.close()

关于STR_TO_DATE函数,如何在字典本身中更新这些值,然后执行insert_many()函数。

for item in temp_data:
    item.update((k, func.STR_TO_DATE(v)) for k,v in item.items() if k == 'time')

你总是可以使用时间模块来测量时间的执行情况

import time
from datetime import datetime
start_time = time.time()
func.STR_TO_DATE('date_value') # datetime.strptime(date_string, "%d %B, %Y")
print('--%s seconds --' %(time.time() - start_time)
    
SVUser
SVUser
发布于 2020-01-19
0 人赞同