如何使用 Python 列表中的 SQLAlchemy 更新 and/or 插入到 MySQL 数据库

Question

我正在构建一个将运行反复插入新数据或根据 ID 更新数据的网络抓取。 if 'id' == 'id': 我的目标是避免重复。 MySQL table 已准备就绪并已构建。使用 SQLAlchemy 在 MySQL 数据库中检查 python 列表之前 inserting/updating 列表的最佳 Pythonic 方法是什么？

以下是我的依赖：

from sqlalchemy.ext.automap import automap_base
from sqlalchemy.orm import Session 
import requests
from bs4 import BeautifulSoup
from time import sleep
from datetime import datetime
import time
engine = create_engine("mysql+pymysql:///blah")

我使用函数从抓取的数据中分配每个 <td>：

def functionscrape( **kwargs ):
    scrape = {
        'id': '',
        'owner': '',
        'street': '',
        'city': '', 
        'state': '', 
        }
    scrape.update(kwargs)
    return (scrape)

下面的列表是一个示例，但会随着每次网络抓取而不断变化。

myList = 
[{
        'id': '111',
        'owner': 'Bob',
        'street': '1212 North',
        'city': 'Anywhere', 
        'state': 'TX', 
        },
{
        'id': '222',
        'owner': 'Mary',
        'street': '333 South',
        'city': 'Overthere', 
        'state': 'AZ', 
        }]

Answer 1

您可以尝试使用 https://marshmallow.readthedocs.io/en/stable/ 库进行验证

构建 Schema 并使用您需要的类型定义字段。您还可以使用 @pre_load 和 @post_load 装饰器来操作您的数据

Answer 2

我正在使用辅助函数来创建动态 sql 更新查询：

def construct_update(table_name, where_vals, update_vals):
    query = table_name.update()
    for k, v in where_vals.items():
        query = query.where(getattr(table_name.c, k) == v)
    return query.values(**update_vals)

基本上，您将 table 和 2 个词典传递给函数。在您的情况下，第一个只是 {'id': id} ，第二个是您要更新的所有值，例如

{
    'owner': 'Bob',
    'street': '1212 North',
    'city': 'Anywhere', 
     etc...
}

辅助函数然后 returns 可以用

执行的查询

my_session = Session(engine)
my_session.execute(query)

不幸的是，使用这种方法，您将不得不单独更新每一行（不能批量更新）——但如果您能接受它，它就可以正常工作

否则这里有一个关于批量更新的类似 post： Bulk update in SQLAlchemy Core using WHERE

如何使用 Python 列表中的 SQLAlchemy 更新 and/or 插入到 MySQL 数据库

How to Update and/or insert to MySQL DB using SQLAlchemy from Python List

python

mysql

sqlalchemy

mysql-python

web-scraping