Skip to main content

A simple python libs

Project description

SimUtil

python容器& libs

1. 安装

pip install simutil

2. 添加配置文件

你需要在你的项目根目录下创建一个.env文件,内容如下:

[environment]                           # 环境变量 环境变量的配置一定要放在[environment]下, 否则读取不到  
ENVIRONMENT = dev                       # 项目环境

#日志配置 
# LOG_DEBUG:日志调试模式  
LOG_DEBUG = true       
# LOG_LEVEL:DEBUG|         
LOG_LEVEL = DEBUG
# 日志路径: 每天产生一个新文件 
# 日志具体文件名: 一直记录到该文件中
LOG_PATH = /Users/**/

# redis配置
REDIS_HOST=127.0.0.1                     
REDIS_PORT=6379                          
REDIS_PASSWORD =                         
REDIS_DB=0                               

# rabbitmq 配置
RABBITMQ_HOST=127.0.0.1
RABBITMQ_PORT=5672
RABBITMQ_USER=test
RABBITMQ_PASSWORD=12345

# oss配置
OSS_ACCESS_DOMAIN=**                     
OSS_ACCESS_KEY=key                       
OSS_ACCESS_SECRET=secret                 
OSS_BUCKET_NAME=bucket_name              
OSS_END_POINT=endpoint

[default.mysql]                         # 数据库等其他配置
DB_TYPE=mysql
DB_HOST=127.0.0.1
DB_PORT=3306
DB_NAME=dev
DB_USER=root
DB_PASSWORD=password
DB_CHARSET=utf8mb4

3. 初始化

容器BASE_PATH注册

from simutil.App import app
import os

base_path = os.path.dirname(os.path.dirname(__file__))
app.register("BASE_PATH", os.path.dirname(os.path.dirname(__file__)))

# 注册完成之后, 你可以在代码任何地方引入app并使用

读取env配置

app('env')('LOG_PATH')                                      # 方法式读取
app('env')('LOG_PATH1', "hahaha")                           # 方法式读取(娶不到舍默认值)
app('env').LOG_PATH                                         # 属性读取
app('env')['LOG_PATH']                                      # 字典读取
app('env').items('default.mysql')                           # 模块读取 
app('env').items('sqlserver', 'DB_HOST')                    # 读取default.mysql中的DB_HOST      
app('env').items('sqlserver', 'DB_HOST1', "123")            # 读取default.mysql中的DB_HOST(带默认值)

log日志

app('log').debug(("status", 0))                             # 记录tuple
app('log').info([1, 2, 3, 4, 5])                            # 记录list
app('log').error({'name':'jemes'})                          # 记录dict
app('log').warning("123444")                                # 记录string    
app('log').critical(5)                                      # 记录number
try:
    a = 100 / 0                                             # error
    raise Exception("i am a Exception")                     # Exception
except Exception as e:
    app('log').info(e)                                      # 记录error or exception

网络请求request

# 同步request get
rep = app('request').get('https://api.longhash.com/', params={}, header={})
text = rep.text
json = rep.json()

# 同步request post
rep = app('request').post('https://api.longhash.com', params={}, header={})
text = rep.text
json = rep.json()

# 异步request get
async def test(url, param={}):
    res = await app('request').aiogett(url, param)
    text = res.text
    json = res.json()
app('async').run([test('https://api.longhash.com/')])              # 异步方法调用

# 异步request post
async def test(url, param={}):
    res = await app('request').aiopost(url, param)
    text = res.text
    json = res.json()
app('async').run([test('https://api.longhash.com/')])              # 异步方法调用

读取项目中配置文件信息

# config 读取app 注册的BASE_PATH下文件配置
app('config')('Config.setting.test_str')                            # {BASE_PATH}/Config/setting.py test_str值
app('config')('Config.setting.test_str', 'default')                 # {BASE_PATH}/Config/setting.py test_str值 取不到则用默认值default
app('config').get('Config.setting.test_str', 'default')             # {BASE_PATH}/Config/setting.py test_str值 字典方式读取

redis

# redis
app('redis').set('name', 123)                                       # 设置key                                 
app('redis').get('name')                                            # 读取key
app('redis').select(1).get('name')                                  # 切换redis DB: 1

# 如果想使用其他非 .env中的配置的redis, 建议使用app.register 注册新的redis实例
from simutil.Redis import Redis
redis = Redis(app('env').REDIS_HOST, app('env').REDIS_PORT)
app.register('redis1', redis)
app('redis1').get('testname')

async方法

# 异步方法
import asyncio
async def hello(index):                                 # 自定义的async方法
    print("number:{}, Hello world!".format(index))
    await asyncio.sleep(1)
    print("end!")
app('async').run([hello(index) for index in range(5)])  # 执行

阿里OSS

# oss
rep = app('request').get('https://api.***.com/index', params={}, header={})

# 上传读取的内容(从内存中)
app('oss').push('oss/api.json', rep.text)                   # oss/api.json为上传到oss的文件路径名称             

# 设置referer
app('oss').referer(["http://api.longhash.com"]).push('test/api.json', rep.text, {'Content-Encoding': 'utf-8'})

# 取消referer 或设置为空
app('oss').referer().push('test/api.json', rep.text, {'Content-Encoding': 'utf-8'})

# 上传本地文件
filepath = /Users/jemes/workspace/simutil/Storage/Logs/20200528.log
app('oss').push_file('test/api.json', filepath, {'Content-Encoding': 'utf-8'})

# rule规则设置
app('oss').rule(allowed_origins=['*'], allowed_methods=['GET'], allowed_headers=['*'], max_age_seconds=100)\
    .push_file('test/api.json', '/Users/jemes/workspace/simutil/Storage/Logs/20200528.log')

Path路径

rep = app('path')                               # 指向BASE_PATH的Pathlib对象
base_path = app('path').resolve()               # = app('BASE_PATH')
config_path = rep.joinpath('Config')            # {BASE_PATH}/Config
logs_path = rep.joinpath('Storage/Logs')        # {BASE_PATH}/Storage/Logs
# 其他操作和Pathlib一致                                             

rabbitMQ

rab = app('rabbitmq')                           # 指向BASE_PATH的Pathlib对象
conn = rab.connection()                         # rabbitmq connection
channel = rab.channel()                         # rabbitmq channel                                           

App容器

# 为了实现代码结偶和复用, 延时加载
# 说明: app('keyname') 拿出来的都是单例, App本身也是单例模式

# 注册class
class DemoA:
    def hello(self):
        return 'DemoA say hello'
demo1 = app.register('demo1', DemoA)                     # 类名注册
demo2 = app.register('demo2', '__main__.DemoA')          # 类路径注册(我在main函数中执行的,所以路径为__main__.DemoA)
demo3 = app.register('demo', DemoA())                    # 实例注册
res = app('demo1').hello()                               # print: DemoA say hello
res = app('demo2').hello()                               # print: DemoA say hello
res = app('demo3').hello()                               # print: DemoA say hello


# 注册function
def add(a, b):
    return a + b
app.register('add', add)                                 # 方法注册
res = app('add')(4, 5)                                   # print: 9

# 注册数据
app.register('listdata', [1, 3, 22, 21, 39])                      
app.register('dictdata', {'name': 'jemes'})
app.register('tupledata', ('name', 'jemes'))
app.register('base_path', str(os.path.dirname(__file__)))
res = app('listdata')                                    # print: [1, 3, 22, 21, 39]
res = app('dictdata')                                    # print: {'name': 'jemes'}
res = app('tupledata')                                   # print: ('name', 'jemes')
res = app('base_path')                                   # print: /Users/jemes/workspace/simutil

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

simutil-1.1.2.tar.gz (13.4 kB view hashes)

Uploaded Source

Built Distribution

simutil-1.1.2-py3-none-any.whl (15.6 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page