selfprivacy-rest-api/selfprivacy_api/graphql/subscriptions/jobs.py

# pylint: disable=too-few-public-methods

from typing import AsyncGenerator, List

from selfprivacy_api.jobs import job_notifications

from selfprivacy_api.graphql.common_types.jobs import ApiJob
from selfprivacy_api.graphql.queries.jobs import get_all_jobs


async def job_updates() -> AsyncGenerator[List[ApiJob], None]:
    # Send the complete list of jobs every time anything gets updated
    async for notification in job_notifications():
        yield get_all_jobs()
feature(jobs): add subscription endpoint 2024-05-15 20:41:36 +00:00			`# pylint: disable=too-few-public-methods`

			`from typing import AsyncGenerator, List`

			`from selfprivacy_api.jobs import job_notifications`

			`from selfprivacy_api.graphql.common_types.jobs import ApiJob`
			`from selfprivacy_api.graphql.queries.jobs import get_all_jobs`


feature(jobs): UNSAFE endpoint to get job updates 2024-05-22 11:04:37 +00:00			`async def job_updates() -> AsyncGenerator[List[ApiJob], None]:`
			`# Send the complete list of jobs every time anything gets updated`
			`async for notification in job_notifications():`
			`yield get_all_jobs()`