Repository¶

class Episode(BaseModel):
    episode_id: int

class Season(BaseModel):
    season: int
    show: Show

class Show(BaseModel):
    show_id: int

class TVTimeUser(BaseModel):
    username: str
    password: str

class UserDB(UserOut):
    hashed_password: str

class UserIn(BaseModel):
    username: str
    password: str

class UserOut(BaseModel):
    user_id: UUID
    username: str
    tv_time_username: str
    disabled: bool | None = None

class TVTimeDataModel(JsonModel):
    username: str = Field(index=True, primary_key=True)
    user_id: Optional[str]
    to_watch: Optional[dict]
    upcoming: Optional[dict]
    profile: Optional[dict]

    class Meta:  # pylint: disable=missing-class-docstring
        global_key_prefix = "tvtime"
        model_key_prefix = "data"

class RedisOMClient:
    def __init__(self) -> None:
        self.redis = get_redis_connection(url=REDIS_URL)
        TVTimeDataModel.Meta.database = self.redis
        # try:
        #     Migrator().run()
        # except MigrationError as e:
        #     logger.error(e)

    def exists(self, key):
        key = f"tvtime:data:{key}"
        return self.redis.exists(key)

    def get_ttl(self, key):
        key = f"tvtime:data:{key}"
        return self.redis.ttl(key)

    def get_tvtime_data(self, key):
        user = TVTimeDataModel.get(key)
        return user

    def get_tvtime_watch_next(self, key):
        data = TVTimeDataModel.get(key)
        if "Watch next" in data.to_watch:
            return data.to_watch["Watch next"]
        else:
            return {}

    def get_tvtime_not_watched_for_while(self, key):
        data = TVTimeDataModel.get(key)
        if "Not watched for a while" in data.to_watch:
            return data.to_watch["Not watched for a while"]
        else:
            return {}

    def get_tvtime_not_started_yet(self, key):
        data = TVTimeDataModel.get(key)
        if "Not started yet" in data.to_watch:
            return data.to_watch["Not started yet"]
        else:
            return {}