Skip to main content

Helo is a simple and small low-level asynchronous ORM using Python asyncio.

Project description

🌎 [English] ∙ [简体中文]

https://img.shields.io/pypi/v/helo.svg https://travis-ci.org/at7h/helo.svg?branch=master https://coveralls.io/repos/github/at7h/helo/badge.svg?branch=master https://app.codacy.com/project/badge/Grade/c68578653eb546488fadddd95f19939c PyPI - Python Version

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+

  • Now only supports MySQL, and the version is 5.7+

  • Integration with web Framework:

  • Not supports table relationship

Quickstart

See the wiki page for more information and quickstart documentation.

Installation

$ pip install helo

See the installation wiki page for more options.

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='g@gmail.com').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 should be:

import quart
import helo

app = quart.Quart(__name__)
app.config["HELO_DATABASE_URL"] = "mysql://user:password@127.0.0.1:3306/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()

Start it:

$ curl http://127.0.0.1:5000/api/authors
[{"email":"g@test.com","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 🤝

  • Special thanks to projects aiomysql and peewee, helo uses aiomysql (as the MySQL connection driver), and referenced peewee in program design.

  • Please feel free to ⭐️ this repository if this project helped you 😉 !

Project details


Download files

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

Source Distributions

No source distribution files available for this release.See tutorial on generating distribution archives.

Built Distribution

helo-0.0.6-py3.7.egg (150.5 kB view hashes)

Uploaded Source

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