文件存储

TXT 文本存储

with open('explore.txt', 'w', encoding='utf-8') as file:
    file.write('\n'.join([question, author, answer]))
    file.write('\n' + '=' * 50 + '\n')

打开方式

在刚才的实例中，open 方法的第二个参数设置成了 a，这样在每次写入文本时不会清空源文件，而是在文件末尾写入新的内容，这是一种文件打开方式。关于文件的打开方式，其实还有其他几种，这里简要介绍一下。

  • r：以只读方式打开文件。文件的指针将会放在文件的开头。这是默认模式。
  • rb：以二进制只读方式打开一个文件。文件指针将会放在文件的开头。
  • r+：以读写方式打开一个文件。文件指针将会放在文件的开头。
  • rb+：以二进制读写方式打开一个文件。文件指针将会放在文件的开头。
  • w：以写入方式打开一个文件。如果该文件已存在，则将其覆盖。如果该文件不存在，则创建新文件。
  • wb：以二进制写入方式打开一个文件。如果该文件已存在，则将其覆盖。如果该文件不存在，则创建新文件。
  • w+：以读写方式打开一个文件。如果该文件已存在，则将其覆盖。如果该文件不存在，则创建新文件。
  • wb+：以二进制读写格式打开一个文件。如果该文件已存在，则将其覆盖。如果该文件不存在，则创建新文件。
  • a：以追加方式打开一个文件。如果该文件已存在，文件指针将会放在文件结尾。也就是说，新的内容将会被写入到已有内容之后。如果该文件不存在，则创建新文件来写入。
  • ab：以二进制追加方式打开一个文件。如果该文件已存在，则文件指针将会放在文件结尾。也就是说，新的内容将会被写入到已有内容之后。如果该文件不存在，则创建新文件来写入。
  • a+：以读写方式打开一个文件。如果该文件已存在，文件指针将会放在文件的结尾。文件打开时会是追加模式。如果该文件不存在，则创建新文件来读写。
  • ab+：以二进制追加方式打开一个文件。如果该文件已存在，则文件指针将会放在文件结尾。如果该文件不存在，则创建新文件用于读写。

JSON 文件存储

读取 JSON

loads 方法将字符串转为 JSON 对象

import json

str = '''
[{
    "name": "Bob",
    "gender": "male",
    "birthday": "1992-10-18"
}, {
    "name": "Selina",
    "gender": "female",
    "birthday": "1995-10-18"
}]
'''
print(type(str))
data = json.loads(str)
print(data)
print(data[0].get('name'))

get 方法还可以传入第二个参数（即默认值）

data[0].get('age')
data[0].get('age', 25)

运行结果如下：

None
25

读取data.json

import json

with open('data.json', 'r') as file:
    str = file.read()
    data = json.loads(str)
    print(data)

输出 JSON

调用 dumps 方法将 JSON 对象转化为字符串,保存 JSON 的格式，可以再加一个参数 indent，代表缩进字符个数

import json

data = [{
    'name': 'Bob',
    'gender': 'male',
    'birthday': '1992-10-18'
}]
with open('data.json', 'w') as file:
    file.write(json.dumps(data, indent=2))

输出文件内容

[
  {
    "name": "Bob",
    "gender": "male",
    "birthday": "1992-10-18"
  }
]

为了输出中文，还需要指定参数 ensure_ascii 为 False，另外还要规定文件输出的编码

with open('data.json', 'w', encoding='utf-8') as file:
    file.write(json.dumps(data, indent=2, ensure_ascii=False))

CSV 文件存储

import csv

with open('data.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['id', 'name', 'age'])
    writer.writerow(['10001', 'Mike', 20])
    writer.writerow(['10002', 'Bob', 22])
    writer.writerow(['10003', 'Jordan', 21])

如果想修改列与列之间的分隔符，可以传入 delimiter 参数

import csv

with open('data.csv', 'w') as csvfile:
    writer = csv.writer(csvfile, delimiter=' ')
    writer.writerow(['id', 'name', 'age'])
    writer.writerow(['10001', 'Mike', 20])
    writer.writerow(['10002', 'Bob', 22])
    writer.writerow(['10003', 'Jordan', 21])

也可以调用 writerows 方法同时写入多行，此时参数就需要为二维列表

import csv

with open('data.csv', 'w') as csvfile:
    writer = csv.writer(csvfile)
    writer.writerow(['id', 'name', 'age'])
    writer.writerows([['10001', 'Mike', 20], ['10002', 'Bob', 22], ['10003', 'Jordan', 21]])

字典的写入方式

import csv

with open('data.csv', 'w') as csvfile:
    fieldnames = ['id', 'name', 'age']
    writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerow({'id': '10001', 'name': 'Mike', 'age': 20})
    writer.writerow({'id': '10002', 'name': 'Bob', 'age': 22})
    writer.writerow({'id': '10003', 'name': 'Jordan', 'age': 21})

读取

import csv  

with open('data.csv', 'r', encoding='utf-8') as csvfile:  
    reader = csv.reader(csvfile)  
    for row in reader:  
        print(row)

关系型数据库存储

创建数据库并且连接

import pymysql

db = pymysql.connect(host='localhost',user='root', password='root', port=3306)
cursor = db.cursor()
cursor.execute('SELECT VERSION()')
data = cursor.fetchone()
print('Database version:', data)
cursor.execute("CREATE DATABASE spiders DEFAULT CHARACTER SET utf8")
db.close()

调用 cursor 方法获得 MySQL 的操作游标，利用游标来执行 SQL 语句。这里我们执行了两句 SQL，直接用 execute 方法执行即可。第一句 SQL 用于获得 MySQL 的当前版本，然后调用 fetchone 方法获得第一条数据

创建表

import pymysql

db = pymysql.connect(host='localhost', user='root', password='123456', port=3306, db='spiders')
cursor = db.cursor()
sql = 'CREATE TABLE IF NOT EXISTS students (id VARCHAR(255) NOT NULL, name VARCHAR(255) NOT NULL, age INT NOT NULL, PRIMARY KEY (id))'
cursor.execute(sql)
db.close()

插入数据

import pymysql

id = '20120001'
user = 'Bob'
age = 20

db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='spiders')
cursor = db.cursor()
sql = 'INSERT INTO students(id, name, age) values(% s, % s, % s)'
try:
    cursor.execute(sql, (id, user, age))
    db.commit()
except:
    db.rollback()
db.close()

需要执行 db 对象的 commit 方法才可实现数据插入，这个方法才是真正将语句提交到数据库执行的方法。对于数据插入、更新、删除操作，都需要调用该方法才能生效。

接下来，我们加了一层异常处理。如果执行失败，则调用 rollback 执行数据回滚

动态插入

import pymysql


db = pymysql.connect(host='localhost', user='root', password='root', port=3306, db='spiders')
cursor=db.cursor
data = {
    'id': '20120001',
    'name': 'Bob',
    'age': 20
}
table = 'students'
keys = ', '.join(data.keys())
values = ', '.join(['% s'] * len(data))
sql = 'INSERT INTO {table}({keys}) VALUES ({values})'.format(table=table, keys=keys, values=values)
try:
   if cursor.execute(sql, tuple(data.values())):
       print('Successful')
       db.commit()
except:
    print('Failed')
    db.rollback()
db.close()

更新数据

若主键不存在就插入，若存在就更新

import pymysql


db = pymysql.connect(host='localhost', user='root', password='root123', port=3306, db='spiders')
cursor=db.cursor()
data = {
    'id': '20120001',
    'name': 'Bob',
    'age': 21
}

table = 'students'
keys = ', '.join(data.keys())
values = ', '.join(['% s'] * len(data))

insert_sql = f'INSERT INTO {table} ({keys}) VALUES ({values})'
update_sql = ', '.join([f"{key} = %s" for key in data])
sql = f'{insert_sql} ON DUPLICATE KEY UPDATE {update_sql}'
params = list(data.values()) * 2
try:
    if cursor.execute(sql, params):
        db.commit()
        print('Successful')
except:
    print('Failed')
    db.rollback()
db.close()

删除数据

table = 'students'
condition = 'age > 20'

sql = 'DELETE FROM  {table} WHERE {condition}'.format(table=table, condition=condition)
try:
    cursor.execute(sql)
    db.commit()
except:
    db.rollback()

db.close()

查询数据

sql = 'SELECT * FROM students WHERE age >= 20'
try:
    cursor.execute(sql)
    print('Count:', cursor.rowcount)
    row = cursor.fetchone()
    while row:
        print('Row:', row)
        row = cursor.fetchone()
except:
    print('Error')

每循环一次，指针就会偏移一条数据

非关系型数据库存储

MongoDB 存储

连接 MongoDB

import pymongo
client = pymongo.MongoClient(host='localhost', port=27017)

指定数据库

db = client.test

指定集合

每个数据库又包含许多集合（collection），它们类似于关系型数据库中的表

collection = db.students

插入数据

import pymongo

# 连接到 MongoDB
client = pymongo.MongoClient(host='localhost', port=27017)

# 选择数据库和集合
db = client.test
collection = db.students

# 定义学生数据
student = {
    'id': '20170101',
    'name': 'Jordan',
    'age': 20,
    'gender': 'male'
}

# 使用 insert_one() 插入单个文档
result = collection.insert_one(student)

# 打印插入结果
print("Inserted student ID:", result.inserted_id)

在 MongoDB 中，每条数据其实都有一个_id 属性来唯一标识。如果没有显式指明该属性，MongoDB 会自动产生一个 ObjectId 类型的_id 属性。insert() 方法会在执行后返回_id 值

插入多条数据

student1 = {
    'id': '20170101',
    'name': 'Jordan',
    'age': 20,
    'gender': 'male'
}

student2 = {
    'id': '20170202',
    'name': 'Mike',
    'age': 21,
    'gender': 'male'
}

result = collection.insert_many([student1, student2])
print(result)
print(result.inserted_ids)

查询

利用 find_one() 或 find() 方法进行查询，其中 find_one() 查询得到的是单个结果，find() 则返回一个生成器对象

result = collection.find_one({'name': 'Mike'})
print(type(result))
print(result)

多条数据查询

results = collection.find({'age': 20})
print(results)
for result in results:
    print(result)

计数

统计查询结果有多少条数据，可以调用 count() 方法

count = collection.count_documents({})
print(count)

排序

排序时，直接调用 sort() 方法，并在其中传入排序的字段及升降序标志即可

results = collection.find().sort('name', pymongo.ASCENDING)
print([result['name'] for result in results])

偏移

在某些情况下，只取某几个元素，这时可以利用 skip() 方法偏移几个位置，比如偏移 2，就忽略前两个元素，得到第三个及以后的元素

results = collection.find().sort('name', pymongo.ASCENDING).skip(2)
print([result['name'] for result in results])

还可以用 limit() 方法指定要取的结果个数

results = collection.find().sort('name', pymongo.ASCENDING).skip(2).limit(2)
print([result['name'] for result in results])

更新

更改一个

condition = {'name': 'Kevin'}
student = collection.find_one(condition)
student['age'] = 26
result = collection.update_one(condition, {'$set': student})
print(result)
print(result.matched_count, result.modified_count)

更改多个

condition = {'age': {'$gt': 20}}
result = collection.update_one(condition, {'$inc': {'age': 1}})
print(result)
print(result.matched_count, result.modified_count)

删除

result = collection.delete_one({'name': 'Kevin'})
print(result)
print(result.deleted_count)
result = collection.delete_many({'age': {'$lt': 25}})
print(result.deleted_count)