Разбираюсь с контекстными вызовами. Столкнулся с ошибкой, которую не могу уже некоторое время побороть.
Написал свой класс:
import psycopg2 from psycopg2 import pool class PgDB: """Класс для работы с БД""" def __init__(self): connection_string = "...тут строка подключения..." try: self.__pool = psycopg2.pool.ThreadedConnectionPool(1, 10, connection_string) self.__conn = self.__pool.getconn() except psycopg2.pool.PoolError as e: self.__pool = None self.__conn = None raise(e) except Exception as e: print(" Unknown error") raise(e) def __enter__(self): return self.__conn def __exit__(self, exc_type, exc_val, exc_tb): if exc_type is not None: self.__conn.rollback() try: self.__conn.commit() except Exception: self.__conn.rollback() finally: self.__conn.close() def execute(self, sqltext='select 1'): ps_cursor = self.__conn.cursor() ps_cursor.execute(sqltext) records = ps_cursor.fetchall() ps_cursor.close() self.__pool.putconn(self.__conn) return records
Вызов вида:
print(PgDB().execute('select * from objects limit 10'))
with PgDB() as db: a = db.execute('select * from objects limit 10')
И вот тут возвращается ошибка
a = db.execute('select * from “objects” limit 10')
AttributeError: ‘psycopg2.extensions.connection’ object has no attribute ‘execute’
с которой у меня разобрать не получилось. Мне не понятно, почему при таком вызове не находится атрибут - метод. Как правильно это реализовать?