feileo/helo

A simple and small low-level asynchronous ORM using Python asyncio.

helo

🌎 [English] ∙ [简体中文]

Helo is a simple and small low-level asynchronous ORM using Python asyncio. It is very intuitive and easy to use.

Helo can help you easily build expressive common SQL statements in your asynchronous applications. You only need to use friendly object-oriented APIs to manipulate data without caring about the details of SQL statement writing and data processing.

• Requires: Python 3.7+
• Only supports MySQL now, and the version is 5.7+
• Integration with web framework:
  • quart, see with quart
• Not supports table relationship now

Installation

$ pip install helo

See the installation wiki page for more options.

Quickstart

See the wiki page for more information and quickstart documentation.

Basic Examples

First, you should to import helo and instantiate a global variable with helo.G

import helo

db = helo.G()

Defining models is simple:

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)

Show some basic examples:

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')
    author_id = await author.save()
    print(author_id)  # 1

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

    await Author.update(email='[email protected]').where(Author.id == author_id).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())

With Quart

If you're using quart , a minimum application example is:

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='[email protected]', password='xxxx'
    ).do()
    author_list = await Author.select().all(False)
    return quart.jsonify(author_list)


app.run()

Run it:

$ curl http://127.0.0.1:5000/api/authors
[{"email":"[email protected]","id":1,"name":"at7h","password":"xxxx"}]

👉 See more examples

Contributing 👏

I hope those who are interested can join in and work together.

Any kind of contribution is expected: report a bug 🐞, give a advice or create a pull request 🙋‍♂️

Thanks 🤝

• Helo used aiomysql and was inspired by peewee in programming. Thank you very much for both!
• Please feel free to ⭐️ this repository if this project helped you 😉 !

helo's Issues

How to update data use pk

Dear auth,
I use this library in my program，and like it. But I make a model using auto_pk settings, but when i want to update one row like this:
eg: await User.update(dict(password=654321, date='2019-01-01'),
User.id == user.id)
It raise ValueError. I found that auto_pk filed id not a type of field type, so User.id == user.id not work, becasue it not Logic_ or Where_ instance. How can i do this? Is there just define id by self not use auto_pk ?