Can't enqueue instance methods

[EliFinkelshteyn]

This is the same issue as #189, which as far as I can tell was never solved. I need to enqueue a class method instead of just a regular function, and I get:

q.enqueue(rq_worker_wrapper.do_something_with_x, x)
*** PicklingError: Can't pickle <type 'function'>: attribute lookup __builtin__.function failed

Is there a recommended way around this?

I really need access to a SQLAlchemy session in the function I want to enqueue, but I can't even work around it by making the function accept that separately:

q.enqueue(some_func, sql_sess, some_data)
*** PicklingError: Can't pickle <class 'sqlalchemy.orm.session.Session'>: it's not the same object as sqlalchemy.orm.session.Session

[JohnSundarraj]

@EliFinkelshteyn , This is how it works.

q.enqueue(MyClass.MyClassMethod,Params)

# MyClass.
class MyClass(object):
  # Initializer.
  def __init__(self):
    self.MySQL = MySQL

  @classmethod
  def MyClassMethod(self,Params):
    Class = self
    self = Class()
    # do something here.

[EliFinkelshteyn]

@JohnSundarraj, thanks for the response! I'm not sure how that answers the question. Here's what I currently have:

class MyClass:
    def __init__(self, config_path):
        self.mysql_connector = MySQLConnector(config_path)

   def do_db_thing(self, some_data):
       sql_sess = self.mysql_connector.get_mysql_connection()
       # do something with sql_sess and some_data here

Using a classmethod here would be incredibly clunky, would require me to basically instantiate a new object every time for no reason, and require me to always know config_path throughout my code. The documentation for rq explicitly says I can use an instance method here, so that's what I was hoping to do. I'm open to other workarounds, but I don't see how I can use a classmethod here in an elegant way, or why I have to use a classmethod in the first place.

[JohnSundarraj]

@EliFinkelshteyn , without creating an instance you cannot use the self.mysql_connector.get_mysql_connection(). Thats y we need to build the instance of that class and use the mysql connection object.

My personal solution for this, is to make db connection class as singleton, so that we don't create new instance everytime.

from sqlalchemy.engine import create_engine
class MySQL(object):
  _Instances = dict()
  def __new__(self):
    if 'Instance' in MySQL._Instances:
      return MySQL._Instances['Instance']
    else:
      self.Engine = create_engine(
        'mysql+mysqldb://User:Password@Host:3306/dbname?charset=utf8&use_unicode=0',
        pool_size=10
      ).connect()
      return super(MySQL,self).__new__(self)

  def __init__(self):
    MySQL._Instances['Instance'] = self

[selwin]

We use Python's pickle to serialize data into Redis and pickle can't realiably pickle instance method. I suggest rewriting your function so that it doesn't take instance method as parameter.

[BenCoDev]

@EliFinkelshteyn I am interested to know how you ended up doing. I am in the same case, and not really eager to rewrite my class.

[yunderboy]

I'm sort of in the situation, the difference is that I can't get rq to pickle a (static) method with @staticmethod annotation. Any suggestions?

[thcrock]

For those looking for a serialization workaround, you can control serializaton and deserialization with Python's __getstate__ and __setstate__. For instance, with the database connection, you can save enough information to reconnect (e.g. the connection URL), but set the database connection itself to None before serializing (in __getstate__). The database connection can be reconstituted in __setstate__.

Example: https://docs.python.org/3/library/pickle.html#pickle-state

I've used this approach with the multiprocessing module in the past. I just tested it with RQ pickling an instance method whose instances has a database connection and it seems to work.

I haven't used static methods, but googling for picklable static methods comes up with a good lead: https://stackoverflow.com/questions/1914261/pickling-a-staticmethod-in-python

[LexxLuey]

Class Nike():
    @job
    task_async.delay(arg):
          #do stuff
   
    def do_it(self, arg):
        self.task_async.delay(arg)

i get this error:

TypeError: Nike.task_async() missing 1 required positional argument: 'arg'

How i instantiate it and call it

n = Nike.do_it(args)