Python 可用于数据库应用程序。与数据交互最常见的是通过数据库管理系统 （DBMS）。
在Python编程环境下可以使用的不同 Python SQL 库，本文介绍 SQLite、MySQL 和 PostgreSQL 数据库进行交互。
整套学习自学教程中应用的数据都是《三國志》、《真·三國無雙》系列游戏中的内容。
通过 Python SQL 库与任何数据仓库交互之前都必须连接到该数据仓库。其中最常用的包括 SQLite、MySQL和PostgreSQL数据库。
创建表单演示 user_list、user_detail 2种表单。
SQLite 是与 Python 应用程序连接的最直接的数据库，是无服务器就可以进行读取和写入数据到文件。不需要安装和运行 SQLite 服务器来执行数据库操作。默认情况下 Python 安装包含一个名为 Python SQL 库 sqlite3 就可以与 SQLite 数据库进行交互。
创建一个链接 SQLite 数据库的方法。
import sqlite3
from sqlite3 import Error
def create_connection(path):
connection = None
try:
connection = sqlite3.connect(path)
print("连接 SQLite DB 成功")
except Error as e:
print(f"错误： '{e}' ")
return connection
sqlite3.connect(path) 返回一个 connection 对象，该对象判断是否连接成功。
创建表单首先要对表单进行连接，连接成功之后才可以进行创建。
connection = create_connection("xxxxx.db")
def execute_query(connection, query):
cursor = connection.cursor()
try:
cursor.execute(query)
connection.commit()
print("连接表单成功")
except Error as e:
print(f"报错 '{e}' ")
创建 user_list 表
create_user_list_table = """
CREATE TABLE IF NOT EXISTS user_list (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
birth INTEGER,
death INTEGER,
gender TEXT,
);
"""
execute_query(connection, create_user_list_table)
创建 user_detail 表
create_user_detail_table = """
CREATE TABLE IF NOT EXISTS user_detail (
id INTEGER PRIMARY KEY AUTOINCREMENT,
weapon TEXT NOT NULL,
character TEXT NOT NULL,
classification TEXT NOT NULL,
user_id INTEGER NOT NULL,
FOREIGN KEY (user_id) REFERENCES user_list (id)
);
"""
execute_query(connection, create_user_detail_table)
使用 INSERT INTO 可以将数据插入到数据表单中。
user_list 表插入数据，id 为自动递增列，因此不需要写入。
create_user_list = """
INSERT INTO
user_list (name, birth , death, gender )
VALUES
('阿会喃', 190, 225, '男'),
('韋昭', 204, 273, '男'),
('伊籍', 162, 226, '男'),
('尹賞', 194, 260, '男'),
('尹大目', 211, 270, '男');
"""
execute_query(connection, create_user_list)
user_detail 表插入数据
create_user_detail = """
INSERT INTO
user_detail (weapon,character, classification,user_id )
VALUES
('刀', '猪突', '武官', 1),
('弓', '豪胆', '文官', 2),
('弓', '冷静', '文官', 3),
('弓', '冷静', '文官', 4),
('弓', '小心', '文官', 5);
"""
execute_query(connection, create_user_detail)
选择数据也称为数据查询，select 作为常用的关键词。
创建一个数据读取的方法
def execute_read_query(connection, query):
cursor = connection.cursor()
result = None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as e:
print(f"错误：'{e}' ")
Selcet 选择数据操作
select_user_list = "SELECT * from user_list"
user_list = execute_read_query(connection, select_user_list)
for user in user_list:
print(user)
(1, '阿会喃', 190, 225, '男')
(2, '韋昭', 204, 273, '男')
(3, '伊籍', 162, 226, '男')
(4, '尹賞', 194, 260, '男')
(5, '尹大目', 211, 270, '男')
Join 数据拼接操作
select_user_list_detail = """
SELECT
user_list.id,
user_list.name,
user_detail.classification
FROM
user_detail
INNER JOIN user_list ON user_list.id = user_detail.user_id
"""
user_list_detail= execute_read_query(connection, select_user_list_detail )
for user_ in user_list_detail:
print(user_)
(1, '阿会喃', '武官')
(2, '韋昭', '文官')
(3, '伊籍', '文官')
(4, '尹賞', '文官')
(5, '尹大目', '文官')
Where 数据条件选择操作
select_user = """
SELECT
user_detail.classification,
COUNT(user_detail.id) as count
FROM
user_list,
user_detail
WHERE
user_list.id = user_detail.user_id
GROUP BY
user_detail.classification
"""
user_count = execute_read_query(connection, select_user )
for count in user_count :
print(count)
('文官', 4)
('武官', 1)
更新数据同样也使用到对应的关键词 UPDATE。
更新前的数据
select_user_detail = "SELECT classification FROM user_detail WHERE id = 2"
user_detail = execute_read_query(connection, select_user_detail)
for user_ in user_detail :
print(user_ )
('文官',)
更新方法和更新后的数据
update_user_detail = """
UPDATE
user_detail
SET
classification = "武官"
WHERE
id = 2
"""
execute_query(connection, update_user_detail )
user_detail= execute_read_query(connection, update_user_detail)
删除数据同样也使用到对应的关键词 DELETE。
delete_detail = "DELETE FROM user_detail WHERE id = 5"
execute_query(connection, delete_detail)
需要使用 mysql-connector-python 为 MySQL 安装 Python SQL 驱动程序，以便从 Python 应用程序中与 MySQL 数据库进行交互。
pip install mysql-connector-python
创建一个链接 MySQL 数据库的方法
import mysql.connector
from mysql.connector import Error
def create_connection(host_name, user_name, user_password):
connection = None
try:
connection = mysql.connector.connect(
host=host_name,
user=user_name,
passwd=user_password
)
print("连接 MySQL DB 成功")
except Error as e:
return connection
connection = create_connection("localhost", "root", "Admin!23456")
执行调用 MySQL 命令连接并进行提交才可以执行。
CREATE DATABASE 创建数据库举例
def create_database(connection, query):
cursor = connection.cursor()
try:
cursor.execute(query)
print("Database 创建成功")
except Error as e:
print(f"错误： '{e}' ")
create_database_query = "CREATE DATABASE DataBaseName"
create_database(connection, create_database_query)
链接创建的新的数据库
def create_connection(host_name, user_name, user_password, db_name):
connection = None
try:
connection = mysql.connector.connect(
host=host_name,
user=user_name,
passwd=user_password,
database=db_name
)
print("连接 MySQL DB 成功")
except Error as e:
print(f"错误： '{e}' ")
return connection
connection = create_connection("localhost", "root", "", "DataBaseName")
创建表单首先要对表单进行连接，连接陈宫之后才可以进行创建。
def execute_query(connection, query):
cursor = connection.cursor()
try:
cursor.execute(query)
connection.commit()
print("连接表单成功")
except Error as e:
print(f"报错 '{e}' ")
创建 user_list 表
create_user_list_table = """
CREATE TABLE IF NOT EXISTS user_list (
id INT AUTO_INCREMENT,
name TEXT NOT NULL,
birth INTEGER,
death INTEGER,
gender TEXT
PRIMARY KEY (id)
) ENGINE = InnoDB
"""
execute_query(connection, create_user_list_table)
创建 user_detail 表
create_user_detail_table= """
CREATE TABLE IF NOT EXISTS user_detail(
id INT AUTO_INCREMENT primary key,
weapon TEXT NOT NULL,
character_ TEXT NOT NULL,
classification TEXT NOT NULL,
user_id INT NOT NULL,
foreign key(user_id) references user_list(id)
) ENGINE = InnoDB
"""
execute_query(connection, create_user_detail_table)
使用 INSERT INTO 和 占位符方法 可以将数据插入到数据表单中。
INSERT INTO 表插入数据
create_user_list = """
INSERT INTO
user_list (name, birth , death, gender )
VALUES
('阿会喃', 190, 225, '男'),
('韋昭', 204, 273, '男'),
('伊籍', 162, 226, '男'),
('尹賞', 194, 260, '男'),
('尹大目', 211, 270, '男')
"""
execute_query(connection, create_user_list)
使用占位符 表插入数据
sql = "INSERT INTO user_detail (weapon,character_, classification,user_id ) VALUES ( %s, %s, %s, %s )"
val = [
('刀', '猪突', '武官', 1),
('弓', '豪胆', '文官', 2),
('弓', '冷静', '文官', 3),
('弓', '冷静', '文官', 4),
('弓', '小心', '文官', 5)
]
cursor = connection.cursor()
cursor.executemany(sql, val)
connection.commit()
选择数据也称为数据查询，select 作为常用的关键词。
创建一个数据读取的方法
def execute_read_query(connection, query):
cursor = connection.cursor()
result = None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as e:
print(f"错误：'{e}' ")
Selcet 选择数据操作
select_user_list = "SELECT * from user_list"
user_list = execute_read_query(connection, select_user_list)
for user in user_list:
print(user)
(1, '阿会喃', 190, 225, '男')
(2, '韋昭', 204, 273, '男')
(3, '伊籍', 162, 226, '男')
(4, '尹賞', 194, 260, '男')
(5, '尹大目', 211, 270, '男')
select_user = """
SELECT
user_detail.classification,
COUNT(user_detail.id) as count
FROM
user_list,
user_detail
WHERE
user_list.id = user_detail.user_id
GROUP BY
user_detail.classification
"""
user_count = execute_read_query(connection, select_user )
for count in user_count :
print(count)
Join 数据拼接操作
select_user_detail = """
SELECT
user_list.id,
user_list.name,
user_detail.classification
FROM
user_detail
INNER JOIN user_list ON user_list.id = user_detail.user_id
"""
user_detail = execute_read_query(connection, select_user_detail)
for user_ in user_detail:
print(user_)
(1, '阿会喃', '武官')
(2, '韋昭', '文官')
(3, '伊籍', '文官')
(4, '尹賞', '文官')
(5, '尹大目', '文官')
Where 数据条件选择操作
select_user_list = "SELECT * from user_list"
user_list = execute_read_query(connection, select_user_list)
for user in user_list:
print(user)
(1, '阿会喃', 190, 225, '男')
(2, '韋昭', 204, 273, '男')
(3, '伊籍', 162, 226, '男')
(4, '尹賞', 194, 260, '男')
(5, '尹大目', 211, 270, '男')
更新数据同样也使用到对应的关键词 UPDATE。
更新前的数据
select_user_list = "SELECT * from user_list"
user_list = execute_read_query(connection, select_user_list)
for user in user_list:
print(user)
(1, '阿会喃', 190, 225, '男')
(2, '韋昭', 204, 273, '男')
(3, '伊籍', 162, 226, '男')
(4, '尹賞', 194, 260, '男')
(5, '尹大目', 211, 270, '男')
更新方法和更新后的数据
update_user_detail = """
UPDATE
user_detail
SET
classification = "武官"
WHERE
id = 2
"""
execute_query(connection, update_user_detail )
user_detail= execute_read_query(connection, update_user_detail)
select_user = """
SELECT
user_detail.classification,
COUNT(user_detail.id) as count
FROM
user_list,
user_detail
WHERE
user_list.id = user_detail.user_id
GROUP BY
user_detail.classification
"""
user_count = execute_read_query(connection, select_user )
for count in user_count :
print(count)
('文官', 3)
('武官', 2)
删除数据同样也使用到对应的关键词 DELETE。
delete_detail = "DELETE FROM user_detail WHERE id = 5"
execute_query(connection, delete_detail)
需要使用 psycopg2 为 PostgreSQL 安装 Python PostgreSQL 驱动程序，以便从 Python 应用程序中与 PostgreSQL 数据库进行交互。
pip install psycopg2
创建一个链接 MySQL 数据库的方法
import psycopg2
from psycopg2 import OperationalError
def create_connection(db_name, db_user, db_password, db_host, db_port):
connection = None
try:
connection = psycopg2.connect(
database=db_name,
user=db_user,
password=db_password,
host=db_host,
port=db_port,
)
print("连接 PostgreSQL DB 成功")
except OperationalError as e:
print(f"错误： '{e}' ")
return connection
connection = create_connection(
"postgres", "xxxx", "xxxx", "localhost", "5432"
)
执行调用 MySQL 命令连接并进行提交才可以执行。
CREATE DATABASE 创建数据库举例
def create_database(connection, query):
connection.autocommit = True
cursor = connection.cursor()
try:
cursor.execute(query)
print("Database 创建成功")
except OperationalError as e:
print(f"错误： '{e}' ")
create_database_query = "CREATE DATABASE DataBaseName"
create_database(connection, create_database_query)
创建表单首先要对表单进行连接，连接成功之后才可以进行创建。
创建 user_list 表
create_user_list_table = """
CREATE TABLE IF NOT EXISTS user_list (
id INT AUTO_INCREMENT,
name TEXT NOT NULL,
birth INTEGER,
death INTEGER,
gender TEXT,
PRIMARY KEY (id)
)
"""
execute_query(connection, create_user_list_table)
创建 user_detail 表
create_user_detail_table = """
CREATE TABLE IF NOT EXISTS user_detail (
id INT AUTO_INCREMENT primary key,
weapon TEXT NOT NULL,
character_ TEXT NOT NULL,
classification TEXT NOT NULL,
user_id references user_list(id)
)
"""
execute_query(connection, create_posts_table)
使用 占位符方法 可以将数据插入到数据表单中。
user_list = [
('阿会喃', 190, 225, '男'),
('韋昭', 204, 273, '男'),
('伊籍', 162, 226, '男'),
('尹賞', 194, 260, '男'),
('尹大目', 211, 270, '男')
]
user_records = ", ".join(["%s"] * len(user_list))
insert_query = (
f"INSERT INTO user_list (name, birth , death, gender) VALUES {user_records}"
)
connection.autocommit = True
cursor = connection.cursor()
cursor.execute(insert_query, user_list)
user_detail = [
('刀', '猪突', '武官', 1),
('弓', '豪胆', '文官', 2),
('弓', '冷静', '文官', 3),
('弓', '冷静', '文官', 4),
('弓', '小心', '文官', 5)
]
user_records = ", ".join(["%s"] * len(user_detail))
insert_query = (
f"INSERT INTO user_detail (weapon,character_, classification,user_id) VALUES {user_records}"
)
connection.autocommit = True
cursor = connection.cursor()
cursor.execute(insert_query, user_records)
选择数据也称为数据查询，select 作为常用的关键词。
选择数据也称为数据查询，select 作为常用的关键词。
创建一个数据读取的方法
def execute_read_query(connection, query):
cursor = connection.cursor()
result = None
try:
cursor.execute(query)
result = cursor.fetchall()
return result
except Error as e:
print(f"错误：'{e}' ")
Selcet 选择数据操作
select_user_list = "SELECT * from user_list"
user_list = execute_read_query(connection, select_user_list)
for user in user_list:
print(user)
('阿会喃', 190, 225, '男'),
('韋昭', 204, 273, '男'),
('伊籍', 162, 226, '男'),
('尹賞', 194, 260, '男'),
('尹大目', 211, 270, '男')
Join 数据拼接操作
select_user_detail= """
SELECT
user_list.id,
user_list.name,
user_detail.classification
FROM
user_detail
INNER JOIN user_list ON user_list.id = user_detail.user_id
"""
user_list_posts = execute_read_query(connection, select_user_detail)
for user_list_post in user_list_posts:
print(user_list_post)
(1, '阿会喃', '武官')
(2, '韋昭', '文官')
(3, '伊籍', '文官')
(4, '尹賞', '文官')
(5, '尹大目', '文官')
Where 数据条件选择操作
select_user_detail = "SELECT classification FROM user_detail WHERE id = 2"
user_detail = execute_read_query(connection, select_user_detail)
for user_ in user_detail :
print(user_ )
('文官',)
更新数据同样也使用到对应的关键词 UPDATE。
更新方法和更新后的数据
update_user_detail = """
UPDATE
user_detail
SET
classification = "武官"
WHERE
id = 2
"""
execute_query(connection, update_user_detail )
user_detail= execute_read_query(connection, update_user_detail)
删除数据同样也使用到对应的关键词 DELETE。
delete_detail = "DELETE FROM user_detail WHERE id = 5"
execute_query(connection, delete_detail)| 留言与评论（共有 0 条评论） “” |
