Help us understand the problem. What is going on with this article?

Python-peewee使用连接方式

方法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连接
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)

参照

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

方法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连接
db = PooledMySQLDatabase('my_database', max_connections=my['max_connections'])

参照

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.

方法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,
)

# 自定义连接类
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连接
db = peewee_async.PooledMySQLDatabase('my_database'
    max_connections=my['max_connections'],
    async_conn=AsyncMySQLConnection)

参照

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

Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
Comments
No comments
Sign up for free and join this conversation.
If you already have a Qiita account
Why do not you register as a user and use Qiita more conveniently?
You need to log in to use this function. Qiita can be used more conveniently after logging in.
You seem to be reading articles frequently this month. Qiita can be used more conveniently after logging in.
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away
ユーザーは見つかりませんでした