Service¶

class TVTimeDataService:
    def __init__(self, redis_client: Annotated[RedisOMClient, Depends()]):
        self.redis = redis_client

    def get_status(self, username):
        status = {
            "exists": self.redis.exists(username),
            "ttl": self.redis.get_ttl(username),
        }
        return status

    def get_all_data(self, username):
        return self.redis.get_tvtime_data(username)

    def get_watch_next(self, username):
        return self.redis.get_tvtime_watch_next(username)

    def get_not_watched_for_while(self, username):
        return self.redis.get_tvtime_not_watched_for_while(username)

    def get_not_started_yet(self, username):
        return self.redis.get_tvtime_not_started_yet(username)

class TVTimeScraperService:
    def __init__(self, redis_client: Annotated[RedisOMClient, Depends()]):
        self.celery = celery
        self.redis = redis_client

    def scrape(self, user: TVTimeUser) -> str:
        logger.debug("Scraping Service Started")
        task = scrape_task.delay(user)
        return task.id

    def get_status(self, task_id):
        logger.debug("Task id {task_id}", task_id=task_id)
        result = AsyncResult(id=task_id, app=self.celery)
        # https://docs.celeryq.dev/en/latest/userguide/tasks.html#built-in-states
        return {"id": result.id, "status": result.status}