Python可以通过使用第三方库如PyMySQL或SQLAlchemy来监听MySQL数据库。这些库提供了API,使得Python程序可以直接与MySQL数据库进行交互,执行查询、更新数据等操作。
在Python中,我们可以使用pymysql
库来连接和操作MySQL数据库,以下是一个简单的例子:
import pymysql 创建连接 conn = pymysql.connect(host='localhost', user='root', password='password', database='test', port=3306) 创建游标 cursor = conn.cursor() 执行SQL语句 cursor.execute('select * from user') 获取所有数据 datas = cursor.fetchall() for data in datas: print(data) 关闭游标和连接 cursor.close() conn.close()
在这个例子中,我们首先导入了pymysql
模块,然后创建了一个到MySQL服务器的连接,我们创建了一个游标对象,通过这个游标,我们可以执行SQL语句并获取结果,记得关闭游标和连接以释放资源。
下面是一个简单的介绍,展示了如何使用Python监听MySQL数据库的相关信息。
pymysql
import pymysql
connection = pymysql.connect(host='localhost', user='root', password='password', database='mydb')
MySQLdb
或pymysql
+SQLAlchemy
MySQLdb
:import MySQLdb
db = MySQLdb.connect(host, user, password, database)
cursor = db.cursor()
cursor.execute("SELECT * FROM my_table")
for row in cursor.fetchall():
print(row)
使用
pymysql
+SQLAlchemy
:from sqlalchemy import create_engine, event
from sqlalchemy.engine import Engine
@event.listens_for(Engine, "connect")
def set_sql_mode(dbapi_connection, connection_record):
cursor = dbapi_connection.cursor()
cursor.execute("SET sql_mode='STRICT_TRANS_TABLES'")
engine = create_engine('mysql+pymysql://user:password@localhost/mydb')
CREATE TRIGGER my_trigger AFTER UPDATE ON my_table FOR EACH ROW BEGIN
触发器逻辑
END;
Python轮询:
import time
while True:
# 检查触发器或表以获取变更
time.sleep(10)
pythonmysqlreplication
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.event import QueryEvent, RotateEvent, FormatDescriptionEvent
stream = BinLogStreamReader(connection_settings={'host': 'localhost', 'port': 3306, 'user': 'root', 'password': 'password'})
for event in stream:
if isinstance(event, QueryEvent):
print(event.query)
请注意,以上代码只是示例,实际使用时需要根据具体场景进行调整,为了安全性,不建议在代码中直接暴露数据库用户名和密码,可以考虑使用环境变量或其他安全方式存储敏感信息。
本文来源于互联网,如若侵权,请联系管理员删除,本文链接:https://www.9969.net/10388.html