Соединение с БД скрипта Python, так как Пул не работает, но работает простое соединение - PullRequest
Новая
       13

Соединение с БД скрипта Python, так как Пул не работает, но работает простое соединение

0 голосов
/ 12 апреля 2019

Я пишу скрипт на Python 3, который прослушивает туннель и сохраняет и обновляет данные в MySQL в зависимости от полученного сообщения.

Я пришел в странное поведение, я установил простое соединение с MySQL, используя модуль pymysql , и все заработало нормально, но через некоторое время это простое соединение закрывается.

Поэтому я решил реализовать подключение к пулу MySQL, и здесь возникает проблема. Что-то происходит без ошибок, но проблема заключается в следующем:

Мой курсор = yield self._pool.execute (запрос, список (filters.values ​​()))

результат курсора = tornado_mysql.pools.Pool объект в 0x0000019DE5D71F98

и такие стеки не делают ничего больше

Если я удаляю , получим из курсора, передайте эту строку, и следующая строка выдаст ошибку

response = yield c.fetchall ()

AttributeError: у объекта 'Future' нет атрибута 'fetchall'

Как я могу исправить соединение с пулом MySQL для правильной работы?

Что я пробовал:

  1. Я использую несколько модулей для подключения к пулу, все идет в одном выпуске

  2. Вернул простое соединение с pymysql и снова заработал

Ниже моего кода:

Файл скрипта Python 

import pika
from model import SyncModel

_model = SyncModel(conf, _server_id)

@coroutine
def main():
    credentials = pika.PlainCredentials('user', 'password')

    try:
        cp = pika.ConnectionParameters(
            host='127.0.0.1',
            port=5671,
            credentials=credentials,
            ssl=False,
        )

        connection = pika.BlockingConnection(cp)
        channel = connection.channel()

        @coroutine
        def callback(ch, method, properties, body):
            if 'messageType' in properties.headers:
                message_type = properties.headers['messageType']
                if message_type in allowed_message_types:
                    result = ptoto_file._reflection.ParseMessage(descriptors[message_type], body)
                    if result:
                        result = protobuf_to_dict(result)
                        if message_type == 'MyMessage':
                            yield _model.message_event(data=result)


                else:
                    print('Message type not in allowed list = ' + str(message_type))
                    print('continue listening...')

        channel.basic_consume(callback, queue='queue', no_ack=True)

        print(' [*] Waiting for messages. To exit press CTRL+C')
        channel.start_consuming()
    except Exception as e:
        print('Could not connect to host 127.0.0.1 on port 5671')
        print(str(e))


if __name__ == '__main__':
    main()

SyncModel 

from tornado_mysql import pools  
from tornado.gen import coroutine, Return
from tornado_mysql.cursors import DictCursor 

class SyncModel(object):
    def __init__(self, conf, server_id):
        self.conf = conf
        servers = [i for i in conf.mysql.servers]

        for s in servers:
            if s['server_id'] == server_id:
                // s hold all data as, host, user, port, autocommit, charset, db, password
                s['cursorclass'] = DictCursor                
                self._pool = pools.Pool(s, max_idle_connections=1, max_recycle_sec=3)

    @coroutine
    def message_event(self, data):
        table_name = 'table_name'
        query = ''
        data = data['message']

        filters = {
            'id': data['id']
        }

        // here the connection fails as describe above
        response = yield self.query_select(table_name, self._pool, filters=filters)


    @coroutine
    def query_select(self, table_name, _pool, filters=None):
        if filters is None:
            filters = {}

        combined_filters = ['`%s` = %%s' % i for i in filters.keys()]
        where = 'WHERE ' + ' AND '.join(combined_filters) if combined_filters else ''
        query = """SELECT * FROM `%s` %s""" % (table_name, where)
        c = self._pool.execute(query, list(filters.values()))

        response = yield c.fetchall()

        raise Return({response})

Весь код работал при простом подключении к базе данных, после того как я начал использовать пул, пример больше не работает. Буду признателен за любую помощь в этом вопросе.

Это отдельный скрипт.

1 Ответ

0 голосов
/ 16 апреля 2019

Подключение к пулу не работает, поэтому переключился обратно на pymysql с двойной проверкой подключения

Я хотел бы опубликовать свой ответ, который работал, у меня работало только это решение

  1. перед подключением к mysql, чтобы проверить, открыто ли соединение, если нет, переподключиться 

    if not self.mysql.open:
    self.mysql.ping(reconnect=True)
...