# Path: pq/api.py



import sys
import os
import tempfile
from pq.api import job
"""Useful Job for the task queue.

Include this file in the ``task_paths`` list if you need them
"""



@job()
async def execute_python(self, code=None):
    """Execute arbitrary python code on a subprocess. For example:

        tasks.queue_task('execute.python', code='print("Hello World!")')
    """
    assert isinstance(code, str), "code must be a string"
    fp, path = tempfile.mkstemp(suffix='.py', text=True)
    try:
        with open(path, 'w') as fp:
            fp.write(code)

        command = '%s %s' % (sys.executable, path)
        result = await self.shell(command)
    finally:
        os.remove(path)
    return result



# Path: pq/api.py



# Path: pq/jobs.py
import sys
import os
import tempfile
from pq.api import job

"""Useful Job for the task queue.

Include this file in the ``task_paths`` list if you need them
"""



@job()
async def execute_python(self, code=None):
    """Execute arbitrary python code on a subprocess. For example:

        tasks.queue_task('execute.python', code='print("Hello World!")')
    """
    assert isinstance(code, str), "code must be a string"
    fp, path = tempfile.mkstemp(suffix='.py', text=True)
    try:
        with open(path, 'w') as fp:
            fp.write(code)

        command = '%s %s' % (sys.executable, path)
        result = await self.shell(command)
    finally:
        os.remove(path)
    return result



@job()