pesto極簡 Python ORM 框架

Python下的極簡orm框架，核心思想，領(lǐng)域?qū)ο?倉庫

1、特色

1、自動化的模型結(jié)構(gòu)映射，不需要復(fù)雜的創(chuàng)建數(shù)據(jù)模型和數(shù)據(jù)庫表結(jié)構(gòu)直接的映射信息，只需要一個(gè)簡單的繼承就可以實(shí)現(xiàn)屬性自動映射

class Example(MysqlBaseModel):
    def __init__(self):
        super(Example, self).__init__(table_name='example')
        
#使用
example= Example()
...
example.save() #對象的增刪改查 輕松實(shí)現(xiàn)

2、超輕量級的配置初始化，只需要在 config.ini 配置對應(yīng)的數(shù)據(jù)庫信息，自動初始化連接，隨時(shí)隨地的執(zhí)行sql

3、簡單實(shí)用的日志工具

# 配置 config.ini
log.path = /opt/logs/pesto-orm/pesto-orm.log
log.level = INFO

# 使用簡單
logger = LoggerFactory.get_logger('dialect.mysql.domain')

4、支持?jǐn)?shù)據(jù)庫事務(wù)

#一個(gè)注解告別，python下的事務(wù)煩惱
@transaction()
def methodX():
    pass

5、環(huán)境隔離的參數(shù)配置工具 config.ini, 公共參數(shù)放default，定制參數(shù)放在各自的環(huán)境

[default]

[dev]

[test]

[prod]

6、等等

2、結(jié)構(gòu)

領(lǐng)域?qū)ο螅侯I(lǐng)域模型對應(yīng)的屬性和行為

倉庫：批量操作領(lǐng)域?qū)ο螅蛘咛厥獾囊恍?shù)據(jù)操作邏輯

領(lǐng)域服務(wù)：統(tǒng)籌領(lǐng)域模型的行為，或者更復(fù)雜的單個(gè)模型無法完成的行為

只需要配置數(shù)據(jù)庫相關(guān)的參數(shù)，通過領(lǐng)域模型，或者倉庫即可操作數(shù)據(jù)，簡單易用，業(yè)務(wù)邏輯復(fù)雜可以加入領(lǐng)域服務(wù)概念

3、示例

pesto-example(flask + pesto-orm) build in python 3.6

add dependencies in requirements(重要):

pesto-orm==0.0.1
mysql-connector-python==8.0.11
Flask==1.0.2

add config in config.ini(重要):

[DEFAULT]
app.key = pesto-orm
log.path = /opt/logs/pesto-orm/pesto-orm.log
log.level = INFO
; db config 目前只支持mysql，歡迎提交其他數(shù)據(jù)庫的實(shí)現(xiàn)
db.database = example
db.raise_on_warnings = True
db.charset = utf8mb4
db.show_sql = True

; profiles config for env
[local]

db.user = root
db.password =
db.host = 127.0.0.1
db.port = 3306

[dev]

[test]

[prod]

run with env(default is local, dev, test, prod)

env=$ENV python ./pesto_example/main.py >> std_out.log 2>&1

main 示例example，可以直接執(zhí)行main方法啟動(需先執(zhí)行數(shù)據(jù)庫的創(chuàng)建，以及配置數(shù)據(jù)的相關(guān)信息)

@app.route('/')
def index():
    data = {'name': 'pesto-example'}
    return jsonify(data)


if __name__ == '__main__':
    port = 8080
    try:
        app.run(host='0.0.0.0', port=port)
    except (KeyboardInterrupt, SystemExit):
        print('')
        logger.info('Program exited.')
    except (Exception,):
        logger.error('Program exited. error info:\n')
        logger.error(traceback.format_exc())
        sys.exit(0)

model 模型創(chuàng)建，只需要配置對應(yīng)的表名和主鍵，領(lǐng)域模型的行為可以擴(kuò)展到該類

class Example(MysqlBaseModel):
    def __init__(self):
        super(Example, self).__init__(table_name='example', primary_key='id')

repository 依賴于模型，執(zhí)行批量或者復(fù)雜的sql邏輯

class ExampleRepository(MysqlBaseRepository):
    def __init__(self):
        super(ExampleRepository, self).__init__(Example)

router api的路由信息，數(shù)據(jù)操作方式

app_example = Blueprint('example', __name__, url_prefix='/examples')

example_repository = ExampleRepository()


@app_example.route('', methods=['GET'])
def examples():
    # 條件查詢
    data = example_repository.query_by(where="")
    if len(data) <= 0:
        jsonify(error="not found any data"), 404

    return jsonify(data)


@app_example.route('/<id>', methods=['GET'])
def example(id):
    # 條件查詢
    data = example_repository.query_first_by(where="`id`= %s", params=(id,))
    if data is None:
        jsonify(error="not found any data"), 404

    return jsonify(data)


@app_example.route('', methods=['POST'])
def save():
    data = request.get_json()
    if data is not None and len(data) > 0:
        example = Example()
        example.set_attrs(data)
        example.created_at = datetime.datetime.now()
        example.updated_at = datetime.datetime.now()
        # 保存數(shù)據(jù)
        example.save()
        return jsonify(example.id)
    else:
        return jsonify(error="not found any data to save"), 400


@app_example.route('/<id>', methods=['DELETE'])
def delete(id):
    result = True
    example = Example()
    example.id = id
    example.deleted_at = datetime.datetime.now()
    # 根據(jù)id刪除數(shù)據(jù)
    example.delete()
    return jsonify(result)


@app_example.route('/<id>', methods=['PUT'])
def update(id):
    result = True
    data = request.get_json()
    example = Example()
    example.set_attrs(data)
    example.id = id
    example.updated_at = datetime.datetime.now()
    # 根據(jù)id更新數(shù)據(jù)
    example.update()
    return jsonify(result)

創(chuàng)建數(shù)據(jù)庫

create database example;

create table example(
    id INT UNSIGNED AUTO_INCREMENT,
    title VARCHAR(255) NOT NULL,
    created_at DATETIME NOT NULL,
    updated_at DATETIME NOT NULL,
    deleted_at DATETIME NOT NULL,
    PRIMARY KEY(id)
    )ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

4、測試

# 查詢?nèi)繑?shù)據(jù)
curl -X GET \
  http://localhost:8080/examples

# 添加一條數(shù)據(jù)
curl -X POST \
  http://localhost:8080/examples \
  -H 'content-type: application/json' \
  -d '{
	"title":"第三個(gè)測試"
}'

# 根據(jù)id查詢
curl -X GET \
  http://localhost:8080/examples/1
  
# 根據(jù)id 更新數(shù)據(jù)
curl -X PUT \
  http://localhost:8080/examples/1 \
  -H 'content-type: application/json' \
  -d '{
	"title":"這是第一個(gè)已改測試"
}'

# 根據(jù)id刪除數(shù)據(jù)
curl -X DELETE \
  http://localhost:8080/examples/3

地址：

https://gitee.com/dreampie/pesto

https://github.com/Dreampie/pesto