python - 使用装饰器连接到 postgres 数据库

Question

我正在开发一个程序，使用 python 和 psycopg2 将我的图片元数据和缩略图存储到 postgres 数据库中。在示例中，我定义了一个类 MyDbase，其中包含创建表、存储值和加载值的方法。这些方法中的每一个都需要连接到数据库和游标对象来执行sql命令。为了避免重复代码来建立连接并获取光标，我创建了一个带有装饰器 connect 的子类 DbDecorators。

我的问题:这是处理此问题的正确方法吗？特别是使用 with 语句并将光标传递给包装器内的 Dbase 方法 (func)？

from functools import wraps
import psycopg2


class MyDbase:
    ''' example using a decorator to connect to a dbase
    '''
    table_name = 'my_table'

    class DbDecorators:
        host = 'localhost'
        db_user = 'db_tester'
        db_user_pw = 'db_tester_pw'
        database = 'my_database'

        @classmethod
        def connect(cls, func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                connect_string = f'host=\'{cls.host}\' dbname=\'{cls.database}\''\
                                 f'user=\'{cls.db_user}\' password=\'{cls.db_user_pw}\''
                result = None
                try:
                    with psycopg2.connect(connect_string) as connection:
                        cursor = connection.cursor()
                        result = func(*args, cursor, **kwargs)

                except psycopg2.Error as error:
                    print(f'error while connect to PostgreSQL {cls.database}: '
                          f'{error}')

                finally:
                    if connection:
                        cursor.close()
                        connection.close()
                        print(f'PostgreSQL connection to {cls.database} is closed')
                return result
            return wrapper

        @staticmethod
        def get_cursor(cursor):
            if cursor:
                return cursor

            else:
                print(f'no connection to database')
                raise()


    @classmethod
    @DbDecorators.connect
    def create_table(cls, *args):
        cursor = cls.DbDecorators().get_cursor(*args)

        sql_string = f'CREATE TABLE {cls.table_name} '\
                     f'(id SERIAL PRIMARY KEY, name VARCHAR(30));'

        print(sql_string)
        cursor.execute(sql_string)

    @classmethod
    @DbDecorators.connect
    def store_value(cls, name, *args):
        cursor = cls.DbDecorators().get_cursor(*args)

        sql_string = f'INSERT INTO {cls.table_name} (name) VALUES (%s);'
        print(sql_string)
        cursor.execute(sql_string, (name,))

    @classmethod
    @DbDecorators.connect
    def load_value(cls, _id, *args):
        cursor = cls.DbDecorators().get_cursor(*args)

        sql_string = f'SELECT * FROM {cls.table_name} where id = \'{_id}\';'
        print(sql_string)
        cursor.execute(sql_string)
        db_row = cursor.fetchone()

        return db_row


def test():
    my_db = MyDbase()
    my_db.create_table()
    my_db.store_value('John Dean')
    db_row = my_db.load_value(1)
    print(f'id: {db_row[0]}, name: {db_row[1]}')

if __name__ == '__main__':
    test()

Answer 1

可能我没有正确收到您的请求。为什么需要装饰器而不使用上下文管理器？就像在任何文件中定义数据库客户端一样，您可以稍后导入它，然后在上下文管理器中使用它 -

from psycopg2 import SomeDataBase  

db = SomeDataBase(credentials)

def create_table(table_name):
        with db:

            sql_string = f'CREATE TABLE {table_name} '\
                     f'(id SERIAL PRIMARY KEY, name VARCHAR(30));'

            db.cursor.execute(sql_string)