时间戳

首页 > 解决方案 > 如何使用 MySQLdb 库通过 python 脚本执行 MySQL 查询？

python - 如何使用 MySQLdb 库通过 python 脚本执行 MySQL 查询？

问题描述

我试图修改 ETL，但我发现老开发人员直接在连接上执行他的命令（ETL 已经运行了几年）。当我尝试自己执行此操作时，出现错误（因为我的编译器希望我从光标执行此操作）。

from etl.utils.logging import info
from etl.mysql.connect import db, db_name
from etl.mysql.operations import add_column_if_not_exists
from etl.utils.array import chunks
from pprint import pprint


def add_column_exclude_from_statistics():
    with db as c:
        # Create new columns where exclude_from_statistics
        info("Creating column exclude from statistics")
        c.execute("""
            UPDATE orders
            INNER JOIN contacts ON orders.id = contacts.`Contact ID`
            IF contacts.`Great Benefactor` = true OR orders.Campaign = `nuit-pour-la-mission` 
                SET orders.exclude_from_statistics = 1
            ELSE 
                SET orders.exclude_from_statistics = 0
            ;
        """)

def main():
    info("Table crm.orders")
    add_column_exclude_from_statistics()


if __name__ == '__main__':
    main()

但它返回'Connection' object has no attribute 'execute'：

(venv) C:\Users\antoi\Documents\Programming\Work\data-tools>py -m etl.task.crm_orders_exclude_from_statistics
2021-06-25 17:12:44.357297 - Connecting to database hozana_data...
2021-06-25 17:12:44.365267 - Connecting to archive database hozana_archive...
2021-06-25 17:12:44.365267 - Table crm.orders
2021-06-25 17:12:44.365267 - Creating column exclude from statistics
Traceback (most recent call last):
  File "C:\Users\antoi\AppData\Local\Programs\Python\Python39\lib\runpy.py", line 197, in _run_module_as_main
    return _run_code(code, main_globals, None,
  File "C:\Users\antoi\AppData\Local\Programs\Python\Python39\lib\runpy.py", line 87, in _run_code
    exec(code, run_globals)
  File "C:\Users\antoi\Documents\Programming\Work\data-tools\etl\task\crm_orders_exclude_from_statistics.py", line 28, in <module>
    main()
  File "C:\Users\antoi\Documents\Programming\Work\data-tools\etl\task\crm_orders_exclude_from_statistics.py", line 24, in main
    add_column_exclude_from_statistics()
  File "C:\Users\antoi\Documents\Programming\Work\data-tools\etl\task\crm_orders_exclude_from_statistics.py", line 12, in add_column_exclude_from_statistic
s
    c.execute("""
AttributeError: 'Connection' object has no attribute 'execute'

这是我们所拥有的etl.mysql.connect

import os
import MySQLdb

from etl.utils.logging import info

db_host = os.environ['DB_HOST']
db_port = int(os.environ['DB_PORT'])
db_user = os.environ['DB_USER']
db_password = os.environ['DB_PASSWORD']
db_name = os.environ['DB_NAME']
db_name_archive = os.environ['DB_ARCHIVE_NAME']

info("Connecting to database {}...".format(db_name))
db = MySQLdb.connect(host=db_host,
                     port=db_port,
                     db=db_name,
                     user=db_user,
                     passwd=db_password)

这样做很奇怪，不是吗？是我的MySQLdb图书馆不是最新的吗？

这是 MySQL 相关的库。我没有找到 MySQLdb：

(venv) C:\Users\antoi\Documents\Programming\Work\data-tools>pip list |findstr mysql
mysql                    0.0.3
mysql-connector-python   8.0.25
mysqlclient              2.0.3

标签： pythonmysqlpython-3.xmysql-python

解决方案

根据文档，您首先需要在连接打开后创建一个游标，因为“连接”对象没有执行方法，但游标有，因此使用您的代码示例：

from etl.utils.logging import info
from etl.mysql.connect import db, db_name
from etl.mysql.operations import add_column_if_not_exists
from etl.utils.array import chunks
from pprint import pprint


def add_column_exclude_from_statistics():
    with db as c:
        # Create new columns where exclude_from_statistics
        info("Creating column exclude from statistics")
        cursor = c.cursor() #Get the cursor
        cursor.execute("""
            UPDATE orders
            INNER JOIN contacts ON orders.id = contacts.`Contact ID`
            IF contacts.`Great Benefactor` = true OR orders.Campaign = `nuit-pour-la-mission` 
                SET orders.exclude_from_statistics = 1
            ELSE 
                SET orders.exclude_from_statistics = 0
            ;
        """)

def main():
    info("Table crm.orders")
    add_column_exclude_from_statistics()


if __name__ == '__main__':
    main()

推荐阅读