README.CN.rst

helo

Helo 是一个简单的小型低级别异步(asyncio) Python ORM。它非常的直观且容易使用。

Helo 可以在你的异步应用中帮助你轻松的构建出富有表达力的常用 SQL 语句，你只需以友好的对象化 API 来操作数据， 而不用关心 SQL 语句编写、数据处理等细节。

• 支持版本: Python 3.7+
• 目前仅支持 MySQL, 版本 5.7+
• 与 web 框架的结合:
  • quart, 参见 quart 项目
• 目前不支持表关系操作

安装

$ pip install helo

更多安装选项请查看 installation 页面。

快速上手

更多信息和 上手文档 请查看 helo wiki 页面。

简单示例

首先，你需要引入 helo 并使用 helo.G 实例化一个全局变量，假定称其为 db:

import helo

db = helo.G()

接下来，声明你的 models:

class Author(helo.Model):
    id = helo.BigAuto()
    name = helo.VarChar(length=45, null=False)
    email = helo.Email(default='')
    password = helo.VarChar(length=100, null=False)
    create_at = helo.Timestamp(default=helo.ON_CREATE)


class Post(helo.Model):
    id = helo.Auto()
    title = helo.VarChar(length=100)
    author = helo.Int(default=0)
    content = helo.Text(encoding=helo.ENCODING.UTF8MB4)
    create_at = helo.Timestamp(default=helo.ON_CREATE)
    update_at = helo.Timestamp(default=helo.ON_UPDATE)

下面的脚本展示一些基本的操作示例:

import asyncio
import datetime


async def show_case():
    # Binding the database(creating a connection pool)
    await db.bind('mysql://user:password@host:port/db')
    # Creating tables
    await db.create_tables([Author, Post])

    # Inserting few rows:

    author = Author(name='at7h', password='1111')
    aid = await author.save()
    print(aid)  # 1

    author = await Author.get(aid)
    print(author.id, author.name)  # 1, at7h

    await Author.update(email='g@gmail.com').where(Author.id == aid).do()

    ret = await Author.insert(name='pope', password='2222').do()
    posts = [
        {'title': 'Python', 'author': 1},
        {'title': 'Golang', 'author': 2},
    ]
    ret = await Post.minsert(posts).do()
    print(ret)  # (2, 1)

    # Supports expressive and composable queries:

    count = await Author.select().count()
    print(count) # 2

    # Last gmail author
    author = await Author.select().where(
        Author.email.endswith('gmail.com')
    ).order_by(
        Author.create_at.desc()
    ).first()
    print(author) # [<Author object at 1>]

    # Using `helo.adict`
    authors = await Author.select(
        Author.id, Author.name
    ).where(
        Author.id < 2
    ).all(wrap=False)
    print(author)  # [{'id': 1, 'name': 'at7h'}]

    # Paginate get authors who wrote Python posts this year
    authors = await Author.select().where(
        Author.id.in_(
            Post.select(Post.author).where(
                Post.update_at > datetime.datetime(2019, 1, 1),
                Post.title.contains('Python')
            ).order_by(
                Post.update_at.desc()
            )
        )
    ).paginate(1, 10)
    print(authors) # [<Author object at 1>]

    # How many posts each author wrote?
    author_posts = await Author.select(
        Author.name, helo.F.COUNT(helo.SQL('1')).as_('posts')
    ).join(
        Post, helo.JOINTYPE.LEFT, on=(Author.id == Post.author)
    ).group_by(
        Author.name
    ).rows(100)


asyncio.run(show_case())

Quart 项目

如果你正在使用 quart, 一个最小的应用示例应该是:

import quart
import helo

app = quart.Quart(__name__)
app.config["HELO_DATABASE_URL"] = "mysql://user:password@host:port/db"

db = helo.G(app)


@app.route('/api/authors')
async def authors():
    await Author.insert(
        name='at7h', email='g@test.com', password='xxxx'
    ).do()
    author_list = await Author.select().all(False)
    return quart.jsonify(author_list)


app.run()

启动此服务:

$ curl http://127.0.0.1:5000/api/authors
[{"email":"g@test.com","id":1,"name":"at7h","password":"xxxx"}]

👉 查看 更多示例

贡献 👏

希望感兴趣的同学可以一起参与，群策群力。

十分欢迎任何类型的贡献： 报 bug 🐞、提 issues 或提交 PR 🙋‍♂️

感谢 🤝

• Helo 使用了 aiomysql ，并在程序设计上受 peewee 启发，非常感谢两者!
• 如果项目对你有帮助请朝 ⭐️ 猛戳 😉 ! 十分感谢!