python-best-practices
2023-04-18 12:35:43 时间
文章目录
导入
新增路径导入
import sys
sys.path.append('project_root_path/')
语句
单行 if else 语句
name = "kangkang" if age >18 else "maria"
字符串
列表拼接为字符串
name_list = ["kangkang", "maria"]
str_name = ",".join(name_list)
# 结果为 str_name = "kangkang,maria"
去除字符串两边空格
raw_name = "test "
name = raw_name.strip()
# 结果为 name = "test"
列表处理
列表倒序
score_list = [91, 92, 78, 88]
reverse_score_list = score_list[::-1]
时间
当地时间
datetime 对象转换成字符串
str_time = datetime_obj.strftime('%Y-%m-%d %H:%M:%S')
字符串转换成 datetime 对象
datetime_obj = datetime.strptime(str_time, '%Y-%m-%d %H:%M:%S')
时间戳转换成字符串
def timestamp_to_str_time(timestamp, _format="%Y-%m-%d %H:%M:%S"):
"""获得字符串时间
:param timestamp: 时间戳
:param _format: 时间格式
:return: 字符串时间
"""
struct_time = time.gmtime(timestamp)
str_time = time.strftime("%Y-%m-%d %H:%M:%S", struct_time)
return str_time
字符串转换成时间戳
def str_time_to_timestamp(str_time, _format='%Y-%m-%d %H:%M:%S'):
"""格式化时间转化为Unix时间戳
:param str_time: 字符串时间
:param _format: 时间格式
:return:
"""
struct_time = time.strptime(str_time, _format)
timestamp = time.mktime(struct_time)
return timestamp
datetime 对象转换成时间戳
from datetime import datetime
datetime.timestamp(datetime_obj)
时间戳转为当地时间字符串
string_time = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(timestamp))
UTC 时间
UTC datetime 对象转换成北京时间字符串
def utc_datetime_obj_to_bj_str_time(utc_datetime_obj):
""" UTC datetime 对象转北京字符串时间
:param utc_datetime_obj: UTC datetime 对象
:return: 北京字符串时间
"""
bj_datetime_obj = utc_datetime_obj + timedelta(hours=8)
bj_str_time = bj_datetime_obj.strftime('%Y-%m-%d %H:%M:%S')
return bj_str_time
UTC 时间字符串转换成北京时间字符串
def utc_str_time_to_bj_str_time(utc_str_time):
""" UTC 字符串时间转北京字符串时间
:param utc_str_time: UTC 字符串时间
:return:
"""
utc_datetime_obj = datetime.strptime(utc_str_time, '%Y-%m-%d %H:%M:%S')
bj_datetime_obj = utc_datetime_obj + timedelta(hours=8)
bj_str_time = bj_datetime_obj.strftime('%Y-%m-%d %H:%M:%S')
return bj_str_time
时间戳转为 UTC 时间字符串
utc_time = datetime.utcfromtimestamp(timestamp)
utc_string_time = utc_time.strftime("%Y-%m-%d %H:%M:%S")
UTC 字符串转为时间戳
import time
from datetime import datetime, timezone
str_time = "2020-06-02 04:58:49"
struct_time = datetime.strptime(str_time, "%Y-%m-%d %H:%M:%S")
utc_struct_time = struct_time.replace(tzinfo=timezone.utc)
timestamp = utc_struct_time.timestamp()
网络请求
POST 请求
api_url = base_url + 'login'
body_json = {
'username': '123',
'password': '123'
}
res_api = requests.post(url=api_url, data=json.dumps(body_json))
GET 请求
api_url = base_url + 'is_login'
params_json = {
'key1': 'value1',
'key2': 'value2'
}
ret = requests.get(url=api_url, params=params_json)
GET 请求封装
使用场景:适用于解析 Json API。 封装特性:异常获取后延迟重试。
import time
import urllib3
from functools import wraps
import requests
from loguru import logger
TIME_OUT_SEC = 120
urllib3.disable_warnings()
logger.add("get_method.log",
format="{time} {level} {message}", level="INFO")
def retry(exception_to_check, tries=3, delay=0.1):
"""
Request an attempt to retry the decorator
Args:
exception_to_check: Exception to check
tries: Number of attempts, default is 3
delay: Pause time, default is 0.1
Returns:
Decorator functions
"""
def deco_retry(f):
@wraps(f)
def f_retry(*args, **kwargs):
mtries = tries
while mtries > 0:
try:
return f(*args, **kwargs)
except exception_to_check as e:
logger.warning(e)
time.sleep(delay)
mtries -= 1
try:
return f(*args, **kwargs)
except exception_to_check as e:
logger.error('Fatal Error: {}'.format(e))
return f_retry
return deco_retry
@retry(exception_to_check=Exception, tries=3, delay=5)
def get_method(url, params=None):
logger.info(f"[get_method] url:{url}, params: {params}")
res = requests.get(url=url, params=params, timeout=TIME_OUT_SEC, verify=False)
if res.status_code == 200:
return res.json()
raise Exception
并发
多线程
with ThreadPoolExecutor(max_workers=2 * multiprocessing.cpu_count() + 1) as executor:
furs = executor.map(fuc_1, data_list)
for fur in furs:
print(fur.exception()) # 用于异常调试,抛出异常。
GET 请求
import asyncio
import aiohttp
from loguru import logger
urllib3.disable_warnings()
class AsyncGetMethod:
def __init__(self, url, params_list):
self.url = url
self.params_list = params_list
self.results = []
self.max_threads = 60
async def handle_tasks(self, work_queue):
while not work_queue.empty():
current_params = await work_queue.get()
try:
await self.get_results(current_params)
except Exception as e:
logger.error('Error for {} {}'.format(current_params, e), exc_info=True)
async def get_results(self, current_params):
html = await self.get_body(current_params)
self.results = self.results + html['transactions']
@retry(ExceptionToCheck=Exception, tries=3, delay=5)
async def get_body(self, current_params):
async with aiohttp.ClientSession() as session:
async with session.get(url=self.url, params=current_params, timeout=120, verify_ssl=False) as response:
return await response.json()
def eventloop(self):
q = asyncio.Queue()
[q.put_nowait(params) for params in self.params_list]
loop = asyncio.new_event_loop()
tasks = [self.handle_tasks(q, ) for task_id in range(self.max_threads)]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
a = AsyncGetMethod(get_api_url,
[{'parameter1': value1, 'parameter2': value2} for value1 in value1_list])
a.eventloop()
print(a.results)
函数
解包传参
def func_1(**property_dic):
print(property_dic)
func_1(field_1=1, field_2=2)
# 输出结果:property_dic={field_1: 1, field_2: 2}
功能
防止脚本重复运行
def exec_cmd(cmd):
"""执行指令的函数
:param cmd: 指令
:return:
"""
res = os.popen(cmd)
res_text = res.read()
res.close()
return res_text
def script_scheduler():
"""脚本调度
调度方案:防止两个监控进程同时在跑
:return:
"""
filename = os.path.basename(__file__)
ps_cmd = "ps -ef|grep " + filename + "|grep -v grep| wc -l"
cmd_res = exec_cmd(ps_cmd)
running_len = int(cmd_res.split('
')[0].strip())
if running_len > 2:
class_log.info('%s is running' % filename)
else:
pass
精确小数
PRECISE_DECIMAL_DIGIT = 4
def precise_decimal(_value):
"""精确小数
:param _value: 需要精确小数的值
:return: 精确小数后的值
"""
return round(_value, PRECISE_DECIMAL_DIGIT)
日志
日志打印
pip3 install loguru
from loguru import logger
# 添加日志配置
logger.add("[日志路径]",
format="{time} {level} {message}", level="INFO")
logger.info("这是一条 info 日志")
# Traceback 记录
@logger.catch
def my_function(x, y, z):
# An error? It's caught anyway!
return 1 / (x + y + z)
单元测试
单元测试保证后续对代码的更改不会引入新的 Bug。
import unittest
class Testxxx(unittest.TestCase):
def test_xxx(self):
self.assertEqual(xxx, True) # 当 xxx 与 True 不一致时报错
self.assertNotEqual(xxx, 0) # 当 xxx 为 0 时报错
数据库交互
ORM(对象-关系映射) - peewee
from peewee import *
# 连接数据库
MYSQL_DB = MySQLDatabase(host=conf.MYSQL_IP,
port=conf.MYSQL_PORT,
user=conf.MYSQL_USERNAME,
passwd=conf.MYSQL_PASSWD,
database=conf.MYSQL_DB)
class BaseModel(Model):
"""基础模型类"""
class Meta:
database = MYSQL_DB
class xxxRecord(BaseModel):
"""xxxRecord 模型类"""
pass
class Meta:
db_table = 'xxx_record'
constraints = [SQL('UNIQUE KEY(field1, field2)')]
# 查询数据
xxx_data = xxxRecord.get(field1=field1_value)
xxx_value = xxx_data.value
文件
打开文件
with open(FILE_PATH + FILE_NAME, 'r') as xxx_file:
for xxx_file_line in xxx_file.readlines():
pass
装饰器
函数运行显示运行日志
import functools
from loguru import logger
def func_run_log(func):
"""函数运行显示运行日志"""
@functools.wraps(func)
def wrapper(*args, **kw):
logger.info('调用函数: %s, 参数: %s' % (func.__name__,
kw))
return func(*args, **kw)
return wrapper
函数报错显示错误跟踪
import functools
import traceback
from loguru import logger
def validate(f):
@functools.wraps(f)
def decorator(*args,**kw):
try:
data = f(*args,**kw)
except Exception as e:
logger.error('发生错误:{}',traceback.format_exc())
data = None
return data
return decorator
常用代码
两数组取交集
def get_intersection(list1, list2):
return [x for x in list1 if x in list2]
get_intersection(["1", "2", "3", "5"], ["2", "4", "5"])
生成 hash
# 生成 xxx_hash_id
raw_string = 字段1 + 字段2 + str(int(time.time()))
bytes_string = bytes(raw_string, encoding="utf8")
xxx_hash_id = hashlib.md5(bytes_string).hexdigest()
执行指令
import os
def execution_command(command):
"""执行指令
:param command: 指令
:return:
"""
res = os.popen(command)
res_text = res.read()
res.close()
return res_text
总结
requests 异常处理
- try…except 处理理连接异常
- 状态码判断 处理理访问异常
- res.text 判断 处理理数据接收异常
- 数据返回状态码判断 处理理 API 数据返回异常
静态检查
- 对代码运行 pylint。
相关文章
- 人工智能 (AI) 正全面改变现代世界
- 为你的团队提供云计算服务的比较
- 道德和治理对人工智能的重要性
- 你的网络AI有你想象得那么智能吗?
- 面向云原生时代 阿里云推出自研“磐久”服务器系列
- 阿里开源玄铁RISC-V系列处理器,推动RISC-V架构走向成熟
- 人工智能创作的一件艺术品获奖引发争论
- IDC:2026年中国AI投资规模或将达266.9亿美元
- 行业专家指出苹果公司需要为企业推出iCloud商务套件
- 人工智能驱动的对话智能如何为金融服务公司提供卓越的客户体验
- 如何打好“黄金 72 小时”里的应急通信攻坚战?
- 告别捅嗓子?AI手机程序通过声音检测新冠,准确率已达到89%
- AI能「踢足球」了,但AI机器人还不能
- 最新AI技术无创解码“大脑语言”,准确率达73%
- 多种边缘集群管理方案对比选型
- 国际教育走进孔孟故乡 百套锐捷智慧黑板助力济南天山外国语新校区落成
- 岂止DALL·E!现在AI画家都会建模、做视频了,以后会什么简直不敢想
- IDC预计今年人工智能收入将接近4500亿美元
- “5G+工业互联网”安全能力及场景化解决方案
- 语言学家重出江湖!从「发音」开始学:这次AI模型要自己教自己