Python3，压箱底的代码片段，提升工作效率稳稳的。

1、引言

小屌丝：鱼哥，这年底了，得不得分享一点压箱底的东西啊
小鱼：… 压箱底的东西，还能分享出来？
小屌丝：唉，为了我能学习更多的知识， 你可以分享分享。
小鱼：确定要压箱底的？
小屌丝：必须的。
小鱼：好吧，既然你这么想要，那就给你。

小屌丝：… 你这…
小鱼：这就是我压箱底的知识啊。
小屌丝：没行到，鱼哥你…

小鱼：算了算了， 我还是跟你分享一点点点点的压箱底的代码吧
小屌丝：这还差不多。

2、代码实例

2.1 操作存储服务

2.1.1 Redis操作

# -*- coding:utf-8 -*-
# @Time   : 2023-12-19
# @Author : Carl_DJ'''
实现功能：对redis链接，写入数据等操作
'''import reids
import requests#链接redis
def redis_con():pool = reids.ConnectPool(host = 'host',port = 8809,decode_responses = True)rd = redis.Redis(connnect_pool = pool)return rdif __name__ == '__main__':#写入redis操作rd = redis_con()rd.set('redis_testDemo1','redis_testDemo2')

2.1.2 MongoDB操作

# -*- coding:utf-8 -*-
# @Time   : 2023-12-19
# @Author : Carl_DJ'''
实现功能：对MongoDB 链接，写入数据等操作
'''from pymongo import MongoClient#链接MongoDB
conn = MongoClient("mongodb://%s:%s@ipaddress:14339/mydb" % ('username', 'password'))
db = conn.mydb
mongo_collection = db.mydata# 批量写入MongoDB操作
res = requests.get(url=url,params=query).json()
commentList = res['data']['commentList']
mongo_collection.insert_many(commentList)

2.1.3 MySQL操作

# -*- coding:utf-8 -*-
# @Time   : 2023-12-19
# @Author : Carl_DJ'''
实现功能：对MySQL链接，查询等等操作
'''import MySQLdb#链接MySQL数据库
db = MySQLdb.connect("localhost","username","password","DBname",charset = 'utf8')# 创建游标
cursor = db.cursor()#执行sql语句
cursor.execute("SELCT * FROM DBname ")# 获取一条数据
data_one = cursor.fetchone()# 获取多条数据
data_all = cursor.fetchall()#关闭数据库链接
db.close()

2.2 异步操作

# -*- coding:utf-8 -*-
# @Time   : 2023-12-19
# @Author : Carl_DJ'''
实现功能：异步编程
'''import aiofilesasync def get_html(session,url):try:async with session.get(url,timeout = 10) as res:if not res.status // 100 == 2:print(f'状态码为：{res.status}')print(f"无法爬取{url},报错了 ")else:res.encoding  = 'utf-8'text = await  res.text()return  textexcept Exception as e:print(f'错误信息:{e}')await  get_html(session,url)async  def download(title_list,content_list):async  with aiofiles.open('{}.txt'.format(title_list[0]),'a',encoding='utf-8') as f:await  f.write('{}'.format(str(content_list)))

2.3 多线程

# -*- coding:utf-8 -*-
# @Time   : 2023-12-19
# @Author : Carl_DJ'''
实现功能：多线程
'''import threading
from datetime import *#自定义全局变量需要的线程数,20
thread_num = 20
#自定义全局变量每个线程需要循环的数量,10
one_work_num = 10
#自定义函数test()
def test():#编写测试代码now = datetime.now()print("打印时间戳：",now)#设置死循环#x =0#while (x ==1):#    print("打印时间戳：",now)
def working():# 引用全局变量global one_work_num#嵌套执行指定循环次数的 test()函数for i in range(0,one_work_num):test()
#自定义thr()函数，来执行多线程
def thr():#自定义一个空的数组，用来存放线程组threads = []#设置循环次数为thread_numfor i in  range(thread_num):#将working()函数放入到线程中t =threading.Thread(target=working,name="T" + str(i))#设置守护线程t.setDaemon(True)threads.append(t)#启动线程并执行for t in threads:t.start()#设置阻塞线程for t in  threads:t.join(2)
if __name__ == "__main__":thr()

3、总结

看到这里，今天的分享差不多就结束了。
其实，在平时工作中，只要善于总结，总会有更高效的代码片段的收集。

关于自动化脚本，可以参照小鱼的这些博文：

• 《Python3，这2个自动化脚本，彻底解放生产力。》
• 《Python3，自从掌握了这3个自动化脚本生成神器，让我的幸福感提升了200%。》
• 《Python3，掌握这4个自动化脚本》
• …等等

也可以关注小鱼的Python专栏：

• 《Python开发实战及代码示例讲解》

我是小鱼：

• CSDN 博客专家；
• 阿里云 专家博主；
• 51CTO博客专家；
• 51认证讲师等；
• 认证金牌面试官；
• 职场面试培训、职业规划师；
• 多个国内主流技术社区的认证专家博主；
• 多款主流产品(阿里云等)测评一、二等奖获得者；

关注小鱼，带你学习更多更专业更前言的Python相关技术。