Discuss / Python / pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'")
Back

pymysql.err.OperationalError: (2003, "Can't connect to MySQL server on '127.0.0.1'")

Topic source

梦_轩昂

#1 Created at ... [Delete] [Delete and Lock User]

在这一节进行测试时，发现创建连接池时，在给loop传为None的是，就会报这个异常。

然后 for 循环时，就会报这个错误。

而创建loop之后，create_pool时就可以正常连接，这么看的话应该不是和安装的包有问题，也应该不是用户的事，不知道哪个大神知道这个问题怎么解决吗？

我是python34，我的代码如下，希望有大神帮忙看一下

loop = asyncio.get_event_loop()

@asyncio.coroutine
def go():
    pool = yield from aiomysql.create_pool(host='127.0.0.1', port=3306,
                                           user='root', password='',
                                           db='mysql', loop=None) # loop为一个实际创建值时就可以成功

    with (yield from pool) as conn:
        cur = yield from conn.cursor()
        yield from cur.execute("SELECT 10")
        print(cur.description)
        (r,) = yield from cur.fetchone()
        assert r == 10
    pool.close()
    yield from pool.wait_closed()

loop.run_until_complete(go())

# 这种方式就是会报错的方式  同时修改中的loop为None
#for x in go():
#    pass

CashQQJ

#2 Created at ... [Delete] [Delete and Lock User]

aiomysql文档里的这段代码有bug。 定义go函数的时候需要把loop参数传进去才正确。

灰_手

#3 Created at ... [Delete] [Delete and Lock User]

这是我写的orm测试，基本上所有方法和参数都被测试到，改了写老师的findNumber和save方法，不过我已经帮修改好了，创建一个文件就可以测试，有兴趣的同学可以看看我改写的orm.py

在此插入代码
import asyncio, logging

from orm import create_pool
from models import User

db_config = {
    'user': 'www-data',
    'password': 'www-data',
    'db': 'awesome'
}

async def test():
    await create_pool(loop, **db_config)

    # 测试count rows语句
    # rows = await User.countRows(), 这句是我改写了老师的findNumber方法，用‘*’比‘id'更高效
    rows = await User.findNumber('*')
    logging.info('rows is: %s' % rows)

    # 测试insert into语句
    if rows < 2:
        for idx in range(5):
            u = User(name='test%s'%(idx), email='test%s@orm.org'%(idx),
                        password='pw', image='/static/img/user.png')
            # rows = await User.countRows('email = ?', u.email)
            rows = await User.countRows('*', email = ?', u.email)
            if rows == 0:
                # await u.register(),这句是我改写了，先加密再调用save方法
                await u.save()
            else:
                print('the email was already registered...')

    # 测试select语句
    users = await User.findAll(orderBy='created_at')
    for user in users:
        logging.info('name: %s, email: %s' % (user.name, user.email))

    # 测试update语句
    user = users[1]
    user.email = 'guest@orm.com'
    user.name = 'guest'
    await user.update()

    # 测试查找指定用户
    test_user = await User.find(user.id)
    logging.info('name: %s, email: %s' % (user.name, user.email))

    # 测试delete语句
    users = await User.findAll(orderBy='created_at', limit=(1, 2))
    for user in users:
        logging.info('delete user: %s' % (user.name))
        await user.remove()

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(test())

灰_手

#4 Created at ... [Delete] [Delete and Lock User]

rows = await User.countRows('*', email = ?', u.email)改成rows = await User.findNumber('*', email = ?', u.email)

  • 1

Reply