Méthode de connexion Python-peewee

Méthode 1

To connect to a MySQL database, we will use MySQLDatabase. After the database name, you can specify arbitrary connection parameters that will be passed back to the driver (either MySQLdb or pymysql).

# http://docs.peewee-orm.com/en/latest/peewee/database.html#using-mysql
from peewee import (
    MySQLDatabase,
    Model,
    DateTimeField,
)
#Connexion MySQL
db = MySQLDatabase('my_database')

class BaseModel(Model):
    """
    DB-Base
    """
    class Meta:
        database = db

    created_at = DateTimeField(default=datetime.datetime.now)
    updated_at = DateTimeField(default=datetime.datetime.now)

    def save(self, *args, **kwargs):
        self.updated_at = datetime.datetime.now()
        return super(BaseModel, self).save(*args, **kwargs)

référence

http://docs.peewee-orm.com/en/latest/index.html

Méthode 2

Connection pooling is provided by the pool module, included in the playhouse extensions library.

The pool supports:

• Timeout after which connections will be recycled.
• Upper bound on the number of open connections.

#http://docs.peewee-orm.com/en/latest/peewee/database.html#connection-pooling

from playhouse.pool import PooledMySQLDatabase
from peewee import (
    Model,
    UUIDField,
    CharField,
)
#Connexion MySQL
db = PooledMySQLDatabase('my_database', max_connections=my['max_connections'])

référence

https://stackoverflow.com/questions/34764266/peewee-pooledmysqldatabase-has-cache-or-buffer Q：peewee PooledMySQLDatabase has cache or buffer? A： No cache, no buffer. It has to do with transaction management. Just be sure that you're running your statements in transactions and you should be fine.

Multi-threaded apps DO NOT REQUIRE A POOL with peewee. Peewee correctly manages per-thread connections even without a pool.

Méthode 3

peewee-async is a library providing asynchronous interface powered by asyncio for peewee ORM.

#https://peewee-async.readthedocs.io/en/latest/
import peewee_async

from peewee import (
    Model,
    UUIDField,
    CharField,
)

#Connexion auto-déterminée
class AsyncMySQLConnection(peewee_async.AsyncMySQLConnection):
    """
    Asynchronous database connection pool.
    """

    def __init__(self, *, database=None, loop=None, timeout=None, **kwargs):
        self.pool = None
        self.loop = loop
        self.database = database
        self.timeout = timeout
        kwargs.setdefault('pool_recycle', 3600)
        self.connect_params = kwargs

#Connexion MySQL
db = peewee_async.PooledMySQLDatabase('my_database'
    max_connections=my['max_connections'],
    async_conn=AsyncMySQLConnection)

référence

https://www.w3cschool.cn/hack_tutorial/hack_tutorial-qay32dz7.html