How to use the arq.Actor function in arq

from arq import Actor, BaseWorker, concurrent


class ActorTest(Actor):
    @concurrent
    async def foo(self, a, b=0):
        with open('foo', 'w') as f:
            r = a + b
            f.write('{}'.format(r))


class Worker(BaseWorker):
    signature = 'foobar'
    shadows = [ActorTest]


class WorkerSignalQuit(Worker):
    """
    worker which simulates receiving sigint after 2 jobs
    """

    @concurrent(Actor.HIGH_QUEUE)
    async def high_add_numbers(self, a, b, c=4):
        with open('high_add_numbers', 'w') as f:
            r = a + b + c
            f.write('{}'.format(r))
        return r

import signal
import time
from pathlib import Path

from arq import Actor, BaseWorker, Job, StopJob, concurrent, cron
from arq.drain import Drain
from arq.testing import MockRedisMixin


class JobConstID(Job):
    @classmethod
    def generate_id(cls, given_id):
        return '__id__'


class DemoActor(Actor):
    job_class = JobConstID

    @concurrent
    async def add_numbers(self, a, b):
        """add_number docs"""
        with open('add_numbers', 'w') as f:
            r = a + b
            f.write('{}'.format(r))

    @concurrent
    async def subtract(self, a, b):
        with open('subtract', 'w') as f:
            try:
                r = a - b
            except TypeError as e:
                r = str(e)

raise StopJob('stopping job with warning', warning=True)


class RealJobActor(DemoActor):
    job_class = Job


class MockRedisDemoActor(MockRedisMixin, DemoActor):
    pass


class Worker(BaseWorker):
    shadows = [DemoActor]


class StartupActor(Actor):
    job_class = JobConstID

    async def startup(self):
        with open('events', 'a') as f:
            f.write('startup[{}],'.format(self.is_shadow))

    @concurrent
    async def concurrent_func(self, v):
        with open('events', 'a') as f:
            f.write('concurrent_func[{}],'.format(v))

    async def shutdown(self):
        with open('events', 'a') as f:
            f.write('shutdown[{}],'.format(self.is_shadow))

class FoobarActor(MockRedisDemoActor):
    name = 'foobar'


def kill_parent():
    time.sleep(0.5)
    os.kill(os.getppid(), signal.SIGTERM)


with Path(__file__).resolve().parent.joinpath('example.py').open() as f:
    EXAMPLE_FILE = f.read()


class ParentActor(MockRedisMixin, Actor):
    v = 'Parent'

    @concurrent
    async def save_value(self, file_name):
        with open(file_name, 'w') as f:
            f.write(self.v)


class ChildActor(ParentActor):
    v = 'Child'


class ParentChildActorWorker(MockRedisMixin, BaseWorker):
    shadows = [ParentActor, ChildActor]

f.write(self.v)


class ChildActor(ParentActor):
    v = 'Child'


class ParentChildActorWorker(MockRedisMixin, BaseWorker):
    shadows = [ParentActor, ChildActor]


class ReEnqueueActor(DemoActor):
    re_enqueue_jobs = True


class CronActor(Actor):
    # using 3:0:0 makes it very unlikely the job will be caused due hitting the right time
    job_class = JobConstID

    @cron(hour=3, minute=0, second=0, run_at_startup=True)
    async def save_foobar(self):
        with open('foobar', 'w') as f:
            f.write(f'foobar the value')

    @cron(hour=3, minute=0, second=0)
    async def save_spam(self):
        with open('spam', 'w') as f:
            f.write(f'spam the value')

    @cron(hour=3, minute=0, second=0, unique=False)
    async def save_not_unique(self):
        with open('not_unique', 'w') as f:

import asyncio
from time import time

import chevron
import uvloop
from aiohttp import web, ClientError, ClientSession
from aiohttp_session import SimpleCookieStorage, get_session
from aiohttp_session import setup as session_setup
from arq import Actor, BaseWorker, RedisSettings, concurrent

R_OUTPUT = 'output'

asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())


class Downloader(Actor):
    re_enqueue_jobs = True

    async def startup(self):
        self.session = ClientSession(loop=self.loop)

    @concurrent
    async def download_content(self, url, count):
        total_size = 0
        errors = []
        start = time()
        for _ in range(count):
            try:
                async with self.session.get(url) as r:
                    content = await r.read()
                    total_size += len(content)
                    if r.status != 200:

from arq import Actor


class FooBar(Actor):
    async def foo(self, a, b, c):
        print(a + b + c)


async def main():
    foobar = FooBar()
    await foobar.enqueue_job('foo', 1, 2, c=48, queue=Actor.LOW_QUEUE)
    await foobar.enqueue_job('foo', 1, 2, c=48)  # this will be queued in DEFAULT_QUEUE
    await foobar.close()

async def main():
    foobar = FooBar()
    await foobar.enqueue_job('foo', 1, 2, c=48, queue=Actor.LOW_QUEUE)
    await foobar.enqueue_job('foo', 1, 2, c=48)  # this will be queued in DEFAULT_QUEUE
    await foobar.close()