python3 实用代码2
所属分类 python
浏览量 13
一、数据处理类
1. collections - 容器数据类型
from collections import Counter, defaultdict, deque, namedtuple, OrderedDict
# Counter - 计数器
cnt = Counter(['a', 'b', 'c', 'a', 'b', 'a'])
print(cnt) # Counter({'a': 3, 'b': 2, 'c': 1})
# defaultdict - 带默认值的字典
dd = defaultdict(list)
dd['key'].append(1) # 无需先检查key是否存在
# namedtuple - 命名元组
Point = namedtuple('Point', ['x', 'y'])
p = Point(1, 2)
print(p.x, p.y)
# deque - 双端队列
dq = deque([1, 2, 3])
dq.appendleft(0) # 左侧添加
dq.pop() # 右侧弹出
2. itertools - 迭代器工具
import itertools
# 排列组合
list(itertools.combinations('ABC', 2)) # [('A','B'), ('A','C'), ('B','C')]
list(itertools.permutations('AB', 2)) # [('A','B'), ('B','A')]
# 无限迭代器
for i in itertools.count(10, 2): # 从10开始,步长为2
if i > 20: break
print(i)
# 循环迭代
for item in itertools.cycle(['A', 'B', 'C']):
# 无限循环
pass
二、文件与目录操作
1. pathlib - 路径操作(Python3.4+推荐)
from pathlib import Path
# 创建Path对象
p = Path('/home/user/documents/file.txt')
# 常用操作
print(p.name) # file.txt
print(p.stem) # file
print(p.suffix) # .txt
print(p.parent) # /home/user/documents
print(p.exists()) # True/False
# 遍历目录
for file in Path('.').glob('*.py'):
print(file)
# 创建目录
Path('new_dir').mkdir(parents=True, exist_ok=True)
# 读写文件
content = p.read_text()
p.write_text('Hello World')
2. os 和 shutil - 操作系统接口
import os, shutil
# 文件操作
os.path.exists('/path/to/file')
os.path.getsize('/path/to/file')
os.path.join('folder', 'file.txt')
# 目录操作
os.listdir('.')
os.makedirs('path/to/dir', exist_ok=True)
# 文件复制、移动、删除
shutil.copy('src.txt', 'dst.txt')
shutil.move('old.txt', 'new.txt')
shutil.rmtree('dir_to_remove') # 递归删除目录
三、日期时间处理
datetime - 日期时间
from datetime import datetime, date, timedelta
import time
# 当前时间
now = datetime.now()
print(now.strftime('%Y-%m-%d %H:%M:%S'))
# 时间计算
tomorrow = now + timedelta(days=1)
last_week = now - timedelta(weeks=1)
# 时间戳
timestamp = time.time()
dt_from_timestamp = datetime.fromtimestamp(timestamp)
# 字符串转时间
dt = datetime.strptime('2023-01-01', '%Y-%m-%d')
四、JSON处理
import json
import csv
# JSON读写
data = {'name': 'John', 'age': 30}
# 写入文件
with open('data.json', 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# 读取文件
with open('data.json', 'r', encoding='utf-8') as f:
loaded_data = json.load(f)
# 字符串转换
json_str = json.dumps(data)
parsed_data = json.loads(json_str)
五、网络请求
requests - HTTP请求
import requests
# GET请求
response = requests.get('https://api.example.com/data')
print(response.status_code)
print(response.json()) # 如果是JSON数据
# 带参数的请求
params = {'key': 'value'}
response = requests.get('https://api.example.com', params=params)
# POST请求
data = {'name': 'John', 'age': 30}
response = requests.post('https://api.example.com', json=data)
# 设置请求头
headers = {'User-Agent': 'my-app'}
response = requests.get(url, headers=headers)
# 会话保持
session = requests.Session()
session.get('https://example.com/login')
六、并发处理
1. concurrent.futures - 线程/进程池
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time
def task(n):
time.sleep(1)
return n * n
# 线程池
with ThreadPoolExecutor(max_workers=5) as executor:
# 提交单个任务
future = executor.submit(task, 5)
print(future.result())
# 批量提交
results = executor.map(task, range(5))
for result in results:
print(result)
# 进程池(适合CPU密集型)
with ProcessPoolExecutor(max_workers=4) as executor:
results = list(executor.map(task, range(10)))
2. asyncio - 异步IO(Python3.5+)
import asyncio
async def fetch_data(url):
# 模拟异步操作
await asyncio.sleep(1)
return f"Data from {url}"
async def main():
# 并发执行多个协程
tasks = [
fetch_data('url1'),
fetch_data('url2'),
fetch_data('url3')
]
results = await asyncio.gather(*tasks)
for result in results:
print(result)
# 运行
asyncio.run(main())
七、日志记录
import logging
# 配置日志
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('app.log'),
logging.StreamHandler()
]
)
# 创建logger
logger = logging.getLogger(__name__)
# 记录日志
logger.debug('Debug message')
logger.info('Info message')
logger.warning('Warning message')
logger.error('Error message')
logger.critical('Critical message')
八、配置管理
configparser - 配置文件
import configparser
# 创建配置
config = configparser.ConfigParser()
config['DEFAULT'] = {'Server': 'localhost', 'Port': '8080'}
config['DATABASE'] = {'Host': 'db.local', 'Name': 'mydb'}
# 写入文件
with open('config.ini', 'w') as f:
config.write(f)
# 读取配置
config.read('config.ini')
server = config['DEFAULT']['Server']
db_host = config['DATABASE']['Host']
九、数据验证
pydantic - 数据验证和设置管理
from pydantic import BaseModel, ValidationError, Field
from typing import List, Optional
from datetime import datetime
class User(BaseModel):
id: int
name: str = Field(..., min_length=2, max_length=50)
email: str
age: Optional[int] = Field(None, ge=0, le=150)
friends: List[int] = []
created_at: datetime = datetime.now()
# 验证数据
try:
user = User(
id=1,
name='John',
email='john@example.com',
age=30,
friends=[2, 3, 4]
)
print(user.dict()) # 转换为字典
except ValidationError as e:
print(e.errors())
十、其他实用工具类
1. 装饰器工具
from functools import wraps, lru_cache
import time
# 缓存装饰器
@lru_cache(maxsize=128)
def expensive_function(n):
time.sleep(2)
return n * n
# 计时装饰器
def timer(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
end = time.time()
print(f"{func.__name__} took {end-start:.2f} seconds")
return result
return wrapper
@timer
def slow_function():
time.sleep(1)
2. 单例模式
class Singleton:
_instance = None
def __new__(cls, *args, **kwargs):
if not cls._instance:
cls._instance = super().__new__(cls)
return cls._instance
def __init__(self, value):
if not hasattr(self, 'initialized'): # 防止重复初始化
self.value = value
self.initialized = True
十一、第三方库推荐
pandas - 数据分析
numpy - 数值计算
Pillow - 图像处理
openpyxl - Excel操作
SQLAlchemy - 数据库ORM
click - 命令行工具
rich - 终端美化
tqdm - 进度条
上一篇
下一篇
N字价格运动与结构
基于深度学习和计算机视觉的K线图交易信号识别
Python f-string 格式化字符串字面量
《走进我的交易室》精华笔记