python的annotate有警告错误，python连接数据库封装

Python连接数据库是项目开发中经常使用的功能之一，通常我们会选择一种或多种数据库进行数据存储和管理。Python提供了多种连接数据库的方式，本文主要介绍Python连接MySQL数据库的封装和实现。

1. Python连接MySQL数据库的基本操作

在Python中，要连接MySQL数据库需要安装Python的MySQL驱动。Python的MySQL连接驱动有多种，比较常用的有 MySQLdb、PyMySQL、mysql-connector-python 等。

（1）使用 MySQLdb 连接MySQL数据库

MySQLdb 是 Python 连接 MySQL 最常用的库，它使用 C 语言编写，速度较快，支持 Python3。

安装 MySQLdb 驱动需要先安装 MySQL C Connector 库：

```

sudo apt-get install libmysqlclient-dev

```

Python 代码示例：

```

import MySQLdb

# 连接数据库

conn = MySQLdb.connect(

host='localhost',

port=3306,

user='root',

passwd='password',

db='test'

)

# 获取游标

cursor = conn.cursor()

# 执行SQL语句

sql = "SELECT * from stu"

cursor.execute(sql)

# 获取所有数据

results = cursor.fetchall()

# 循环遍历输出每条记录的结果

for row in results:

print(row)

# 关闭游标和连接

cursor.close()

conn.close()

```

上面的代码中，我们通过 MySQLdb.connect() 方法连接 MySQL 数据库，并通过 conn.cursor() 方法获取游标，使用 cursor.execute() 执行 SQL 语句，最后通过 cursor.fetchall() 获取查询结果。

（2）使用 PyMySQL 连接MySQL数据库

PyMySQL 是一个纯 Python 编写的 MySQL 客户端库，相对于 MySQLdb 来说，它更加稳定且支持 Python3。

Python 代码示例：

```

import pymysql

# 连接数据库

conn = pymysql.connect(

host='localhost',

port=3306,

user='root',

passwd='password',

db='test'

)

# 获取游标

cursor = conn.cursor()

# 执行SQL语句

sql = "SELECT * from stu"

cursor.execute(sql)

# 获取所有数据

results = cursor.fetchall()

# 循环遍历输出每条记录的结果

for row in results:

print(row)

# 关闭游标和连接

cursor.close()

conn.close()

```

上面的代码中，我们通过 pymysql.connect() 方法连接 MySQL 数据库，通过 conn.cursor() 方法获取游标，使用 cursor.execute() 执行 SQL 语句，最后通过 cursor.fetchall() 获取查询结果。

2. Python连接MySQL数据库的封装

减少冗余代码，提高代码可读性和维护性，我们可以封装 Python 连接 MySQL 数据库的代码，使其更加简洁易用。

（1）使用 MySQLdb 封装

```

import MySQLdb

class MysqlClient(object):

def __init__(self, host, user, passwd, db):

self.host = host

self.user = user

self.passwd = passwd

self.db = db

def _getConnection(self):

self.conn = MySQLdb.connect(

host=self.host,

user=self.user,

passwd=self.passwd,

db=self.db

)

self.conn.autocommit(True)

def _closeConnection(self):

self.conn.close()

def executeCommand(self, sql):

self._getConnection()

cursor = self.conn.cursor()

try:

cursor.execute(sql)

results = cursor.fetchall()

return results

finally:

cursor.close()

self._closeConnection()

```

上面的代码中，我们定义了一个 MysqlClient 类，该类包含了获取数据库连接、关闭数据库连接和执行 SQL 语句的方法，可支持多线程，并保持连接池。

（2）使用 PyMySQL 封装

```

import pymysql

class MySQLClient():

def __init__(self, host, user, password, database):

self.host = host

self.user = user

self.password = password

self.database = database

def connect(self):

'''

连接数据库

'''

self.conn = pymysql.connect(

host=self.host,

user=self.user,

password=self.password,

database=self.database,

charset='utf8',

cursorclass=pymysql.cursors.DictCursor,

autocommit=True

)

def execute(self, sql, params=None):

'''

执行查询语句

'''

self.connect()

with self.conn.cursor() as cursor:

cursor.execute(sql, params)

result = cursor.fetchall()

return result

def update(self, sql, params=None):

'''

执行更新操作

'''

self.connect()

with self.conn.cursor() as cursor:

cursor.execute(sql, params)

self.conn.commit()

```

上面的代码中，我们定义了一个 MySQLClient 类，该类包含了连接 MySQL 数据库、执行查询语句和执行更新操作的方法。

3. Python连接MySQL数据库的性能优化

在高并发场景下，Python连接 MySQL 数据库的性能往往是瓶颈。下面介绍几个优化方式：

（1）使用连接池

为了避免多次建立连接，可以使用连接池复用连接。Python中，我们可以使用第三方库来实现连接池，比如：DBUtils、PySQLPool 和 SQLAlchemy 等。

我们以 DBUtils 为例：

```

import psycopg2

from dbutils.pooled_db import PooledDB

class Database:

def __init__(self, *args, **kwargs):

self.pool = PooledDB(

creator=psycopg2, # 数据库驱动名称

maxconnections=6, # 连接池允许的最大连接数，-1 无限制

mincached=2, # 初始化时连接池中至少创建的空闲连接，0 不创建

maxcached=5, # 连接池中最多闲置的连接，超过的连接将被关闭并删除

blocking=True, # 连接池达到最大值后是否阻塞等待连接释放

*args,

**kwargs

)

def execute(self, sql, *args):

conn = self.pool.connection()

cur = conn.cursor()

cur.execute(sql, args)

result = cur.fetchall()

cur.close()

conn.close()

return result

```

（2）使用 ORM 框架

Python 中的 ORM 框架可以将数据库操作映射为对象操作，进一步简化 SQL 操作。使用 ORM 框架能够更好地管理和组织代码，提高编码效率和可读性。

比较常用的 Python ORM 框架有 Django ORM、SQLAlchemy 等，我们以 SQLAlchemy 为例：

```

from sqlalchemy import create_engine, Column, Integer, String, DateTime, Text, Boolean

from sqlalchemy.ext.declarative import declarative_base

from sqlalchemy.orm import sessionmaker

from datetime import datetime

from sqlalchemy.sql import text

Base = declarative_base()

class Book(Base):

__tablename__ = 'book'

id = Column(Integer, primary_key=True)

title = Column(String(200), nullable=False)

author = Column(String(50), nullable=False)

create_at = Column(DateTime, default=datetime.now)

def __repr__(self):

return f""

class Database:

def __init__(self, *args, **kwargs):

self.engine = create_engine(*args, **kwargs)

self.session = sessionmaker(bind=self.engine)

def execute(self, sql, *args):

with self.session() as db:

result = db.execute(text(sql), *args)

return result.fetchall()

def add(self, table, **kwargs):

with self.session() as db:

item = table(**kwargs)

db.add(item)

db.commit()

def delete(self, table, id):

with self.session() as db:

item = db.query(table).get(id)

db.delete(item)

db.commit()

def update(self, table, id, **kwargs):

with self.session() as db:

item = db.query(table).get(id)

for key, value in kwargs.items():

setattr(item, key, value)

db.commit()

```

上面的代码中，我们使用 SQLAlchemy 实现了一个 Book 表的基本操作，包括添加、删除和更新操作。

结论

本文中，我们介绍了 Python 连接 MySQL 数据库的基本操作、封装和性能优化。连接 MySQL 数据库是项目开发中必不可少的功能之一，我们可以根据项目需求，选择相应的驱动和优化方式，提高代码的效率和可读性。 如果你喜欢我们三七知识分享网站的文章， 欢迎您分享或收藏知识分享网站文章 欢迎您到我们的网站逛逛喔！https://www.ynyuzhu.com/