基于__call__的类装饰器实现SQLAlchemy连接关闭遇参数缺失报错

Got it, let's break down why you're hitting that TypeError: __call__() missing 1 required positional argument: 'args' and fix your decorator to properly handle closing SQLAlchemy connections after queries.

What's Causing the Error? #

When you decorate an instance method like execute_query, Python automatically passes the instance (self) as the first argument when the method is called. Your decorator's __call__ method only expects a single args parameter, which doesn't account for this implicit self argument. Additionally, your current setup doesn't handle variable arguments or keyword arguments that your query method might need later.

Corrected Code with Proper Connection Handling #

Here's a revised version of your code that fixes the error and adds real SQLAlchemy connection-closing logic:

from sqlalchemy import create_engine

class Connection(object):
    class close_sqlalchemy_connection(object):
        def __init__(self, f):
            print("inside close_sqlalchemy_connection.__init__()")
            self.f = f

        def __call__(self, *args, **kwargs):
            print("inside close_sqlalchemy_connection.__call__()")
            # Execute the original query method with all passed arguments
            result = self.f(*args, **kwargs)
            
            # Get the Connection instance from args (first argument is self)
            db_instance = args[0]
            # Close the SQLAlchemy connection if it exists
            if hasattr(db_instance, 'mysql_connection') and not db_instance.mysql_connection.closed:
                db_instance.mysql_connection.close()
                print("SQLAlchemy connection closed successfully")
            
            return result

    def __init__(self, db_url):
        # Initialize SQLAlchemy engine and connection
        self.engine = create_engine(db_url)
        self.mysql_connection = self.engine.connect()
        print("SQLAlchemy connection established")

    @close_sqlalchemy_connection
    def execute_query(self, query):
        print('Executing query...')
        result = self.mysql_connection.execute(query)
        print(f"Query completed, fetched {result.rowcount} rows")
        return result.fetchall()

# Example usage
if __name__ == "__main__":
    # Replace with your MySQL connection string (format: mysql+pymysql://user:password@host/dbname)
    db_url = "mysql+pymysql://your_user:your_password@localhost/your_database"
    db = Connection(db_url)
    # Run a sample query
    data = db.execute_query("SELECT * FROM your_table LIMIT 5")
    print("Query results:", data)

Key Changes Explained #

1. Updated __call__ Method:

   • Changed to accept *args and **kwargs to handle any number of positional/keyword arguments passed to the decorated method, including the implicit self argument from the instance method.
   • Calls the original function with *args and **kwargs to ensure all parameters are passed correctly.

2. Real Connection Management:

   • The Connection class now takes a database URL and initializes a real SQLAlchemy engine and connection.
   • The decorator accesses the instance via args[0] (the first argument is always self for instance methods) to close the connection after the query runs.

3. Robust Closing Logic:

   • Checks if the connection exists and isn't already closed before attempting to close it, preventing unnecessary errors.

Alternative Note #

While a decorator works here, SQLAlchemy also supports context managers for connection handling, which is often more idiomatic:

with self.engine.connect() as conn:
    result = conn.execute(query)
# Connection is automatically closed when exiting the with block

But since you specifically wanted to use a class decorator, the code above should meet your needs perfectly.

内容的提问来源于stack exchange，提问作者user1050619