refactor(redis): switch to redis-py

We have to vendor redis-py typing because types-redis has an incomplete
redis.asyncio module typing.

I proposed my typing here: python/typeshed#7820

The change also makes mypy discovers some un-catched typing issue.

Author: sileht

mergify_engine/context.py

@@ -620,7 +620,7 @@ async def clear_user_permission_cache_for_user(
     ) -> None:
         await redis.hdel(
             cls._users_permission_cache_key_for_repo(owner["id"], repo["id"]),
-            user["id"],
+            str(user["id"]),
         )
 
     @classmethod
@@ -655,7 +655,7 @@ async def get_user_permission(
             key = self._users_permission_cache_key
             cached_permission = typing.cast(
                 typing.Optional[github_types.GitHubRepositoryPermission],
-                await self.installation.redis.cache.hget(key, user["id"]),
+                await self.installation.redis.cache.hget(key, str(user["id"])),
             )
             if cached_permission is None:
                 permission = typing.cast(

mergify_engine/count_seats.py

@@ -181,7 +181,7 @@ def _add_user(user: github_types.GitHubAccount) -> None:
     transaction = await redis_cache.pipeline()
     for user_id, user_login in users.items():
         user_key = f"{user_id}~{user_login}"
-        await transaction.zadd(repo_key, **{user_key: time.time()})
+        await transaction.zadd(repo_key, {user_key: time.time()})
 
     await transaction.execute()
 
@@ -425,7 +425,7 @@ async def count_and_send(redis_cache: redis_utils.RedisCache) -> None:
 
 
 async def report(args: argparse.Namespace) -> None:
-    redis_links = redis_utils.RedisLinks()
+    redis_links = redis_utils.RedisLinks(name="report")
     if args.daemon:
         service.setup("count-seats")
         await count_and_send(redis_links.cache)

mergify_engine/debug.py

@@ -77,10 +77,13 @@ async def report_dashboard_synchro(
     )
 
 
-async def report_worker_status(owner: github_types.GitHubLogin) -> None:
+async def report_worker_status(
+    redis_links: redis_utils.RedisLinks, owner: github_types.GitHubLogin
+) -> None:
     stream_name = f"stream~{owner}".encode()
-    redis_links = redis_utils.RedisLinks()
-    streams = await redis_links.stream.zrangebyscore(
+    streams: typing.List[
+        typing.Tuple[bytes, float]
+    ] = await redis_links.stream.zrangebyscore(
         "streams", min=0, max="+inf", withscores=True
     )
 
@@ -91,9 +94,13 @@ async def report_worker_status(owner: github_types.GitHubLogin) -> None:
         print("* WORKER: Installation not queued to process")
         return
 
-    planned = datetime.datetime.utcfromtimestamp(streams[pos]).isoformat()
+    planned = datetime.datetime.utcfromtimestamp(streams[pos][1]).isoformat()
 
-    attempts = await redis_links.stream.hget("attempts", stream_name) or 0
+    attempts_raw = await redis_links.stream.hget("attempts", stream_name)
+    if attempts_raw is None:
+        attempts = 0
+    else:
+        attempts = int(attempts)
     print(
         "* WORKER: Installation queued, "
         f" pos: {pos}/{len(streams)},"
@@ -170,7 +177,7 @@ def _url_parser(
 async def report(
     url: str,
 ) -> typing.Union[context.Context, github.AsyncGithubInstallationClient, None]:
-    redis_links = redis_utils.RedisLinks(max_idle_time=0)
+    redis_links = redis_utils.RedisLinks(name="debug")
 
     try:
         owner_login, repo, pull_number = _url_parser(url)
@@ -234,7 +241,7 @@ async def report(
         installation.installation["id"], db_sub, db_tokens, "DASHBOARD", slug
     )
 
-    await report_worker_status(owner_login)
+    await report_worker_status(redis_links, owner_login)
 
     if repo is not None:
         repository = await installation.get_repository_by_name(repo)
@@ -281,6 +288,7 @@ async def report(
                 )
             except http.HTTPNotFound:
                 print(f"Pull request `{url}` does not exist")
+                await redis_links.shutdown_all()
                 return client
 
             # FIXME queues could also be printed if no pull number given
@@ -319,9 +327,10 @@ async def report(
                 )
                 print(f"[Summary]: success | {summary_title}")
                 print("> " + "\n> ".join(summary.strip().split("\n")))
-
+            await redis_links.shutdown_all()
             return ctxt
 
+    await redis_links.shutdown_all()
     return client
 
 

mergify_engine/delayed_refresh.py

@@ -61,7 +61,7 @@ async def _set_current_refresh_datetime(
 ) -> None:
     await repository.installation.redis.cache.zadd(
         DELAYED_REFRESH_KEY,
-        **{_redis_key(repository, pull_number): at.timestamp()},
+        {_redis_key(repository, pull_number): at.timestamp()},
     )
 
 
@@ -129,16 +129,18 @@ async def send(
     for subkey in keys:
         (
             owner_id_str,
-            owner_login,
+            owner_login_str,
             repository_id_str,
-            repository_name,
+            repository_name_str,
             pull_request_number_str,
         ) = subkey.split("~")
         owner_id = github_types.GitHubAccountIdType(int(owner_id_str))
         repository_id = github_types.GitHubRepositoryIdType(int(repository_id_str))
         pull_request_number = github_types.GitHubPullRequestNumber(
             int(pull_request_number_str)
         )
+        repository_name = github_types.GitHubRepositoryName(repository_name_str)
+        owner_login = github_types.GitHubLogin(owner_login_str)
 
         LOG.info(
             "sending delayed pull request refresh",

mergify_engine/exceptions.py

@@ -15,7 +15,7 @@
 import datetime
 import typing
 
-import yaaredis
+from redis import exceptions as redis_exceptions
 
 from mergify_engine.clients import http
 
@@ -118,11 +118,11 @@ def need_retry(
         elif exception.response.status_code == 403:
             return datetime.timedelta(minutes=3)
 
-    elif isinstance(exception, yaaredis.exceptions.ResponseError):
+    elif isinstance(exception, redis_exceptions.ResponseError):
         # Redis script bug or OOM
         return datetime.timedelta(minutes=1)
 
-    elif isinstance(exception, yaaredis.exceptions.ConnectionError):
+    elif isinstance(exception, redis_exceptions.ConnectionError):
         # Redis down
         return datetime.timedelta(minutes=1)
 

mergify_engine/migrations/__init__.py

@@ -33,14 +33,12 @@ async def run(redis_cache: redis_utils.RedisCache) -> None:
     await _run_scripts("cache", redis_cache)
 
 
-async def _run_scripts(
-    dirname: str, redis: typing.Union[redis_utils.RedisCache, redis_utils.RedisStream]
-) -> None:
-    current_version = await redis.get(MIGRATION_STAMPS_KEY)
-    if current_version is None:
+async def _run_scripts(dirname: str, redis: redis_utils.RedisCache) -> None:
+    current_version_raw: typing.Optional[str] = await redis.get(MIGRATION_STAMPS_KEY)
+    if current_version_raw is None:
         current_version = 0
     else:
-        current_version = int(current_version)
+        current_version = int(current_version_raw)
 
     files = pkg_resources.resource_listdir(__name__, dirname)
     for script in sorted(files):

mergify_engine/queue/merge_train.py

@@ -1354,7 +1354,7 @@ async def iter_trains(
             await train.load(train_raw)
             yield train
 
-    async def load(self, train_raw: typing.Optional[bytes] = None) -> None:
+    async def load(self, train_raw: typing.Optional[str] = None) -> None:
         if train_raw is None:
             train_raw = await self.repository.installation.redis.cache.hget(
                 self._get_redis_key(), self._get_redis_hash_key()