Python-Peewee-Verbindungsmethode

Methode 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,
)
#MySQL-Verbindung
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)

Referenz

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

Methode 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,
)
#MySQL-Verbindung
db = PooledMySQLDatabase('my_database', max_connections=my['max_connections'])

Referenz

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.

Methode 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,
)

#Selbstbestimmte Verbindung
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

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

Referenz

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