Add robot account management

reconcile/cli.py

@@ -1962,6 +1962,14 @@ def quay_membership(ctx: click.Context) -> None:
     run_integration(reconcile.quay_membership, ctx)
 
 
+@integration.command(short_help="Manages robot accounts in Quay organizations.")
+@click.pass_context
+def quay_robot_accounts(ctx: click.Context) -> None:
+    import reconcile.quay_robot_accounts
+
+    run_integration(reconcile.quay_robot_accounts, ctx)
+
+
 @integration.command(short_help="Mirrors external images into GCP Artifact Registry.")
 @click.pass_context
 @binary(["skopeo"])

reconcile/gql_definitions/quay_robot_accounts/quay_robot_accounts.gql

@@ -0,0 +1,25 @@
+# qenerate: plugin=pydantic_v1
+
+query QuayRobotAccounts {
+  robot_accounts: quay_robots_v1 {
+    name
+    description
+    quay_org {
+      name
+      instance {
+        name
+        url
+      }
+      automationToken {
+        path
+        field
+        version
+      }
+    }
+    teams
+    repositories {
+      name
+      permission
+    }
+  }
+}

reconcile/gql_definitions/quay_robot_accounts/quay_robot_accounts.py

@@ -0,0 +1,104 @@
+"""
+Generated by qenerate plugin=pydantic_v1. DO NOT MODIFY MANUALLY!
+"""
+from collections.abc import Callable  # noqa: F401 # pylint: disable=W0611
+from datetime import datetime  # noqa: F401 # pylint: disable=W0611
+from enum import Enum  # noqa: F401 # pylint: disable=W0611
+from typing import (  # noqa: F401 # pylint: disable=W0611
+    Any,
+    Optional,
+    Union,
+)
+
+from pydantic import (  # noqa: F401 # pylint: disable=W0611
+    BaseModel,
+    Extra,
+    Field,
+    Json,
+)
+
+
+DEFINITION = """
+query QuayRobotAccounts {
+  robot_accounts: quay_robots_v1 {
+    name
+    description
+    quay_org {
+      name
+      instance {
+        name
+        url
+      }
+      automationToken {
+        path
+        field
+        version
+      }
+    }
+    teams
+    repositories {
+      name
+      permission
+    }
+  }
+}
+"""
+
+
+class ConfiguredBaseModel(BaseModel):
+    class Config:
+        smart_union=True
+        extra=Extra.forbid
+
+
+class QuayInstanceV1(ConfiguredBaseModel):
+    name: str = Field(..., alias="name")
+    url: str = Field(..., alias="url")
+
+
+class VaultSecretV1(ConfiguredBaseModel):
+    path: str = Field(..., alias="path")
+    field: str = Field(..., alias="field")
+    version: Optional[int] = Field(..., alias="version")
+
+
+class QuayOrgV1(ConfiguredBaseModel):
+    name: str = Field(..., alias="name")
+    instance: QuayInstanceV1 = Field(..., alias="instance")
+    automation_token: Optional[VaultSecretV1] = Field(..., alias="automationToken")
+
+
+class QuayRepositoryV1(ConfiguredBaseModel):
+    name: str = Field(..., alias="name")
+    permission: str = Field(..., alias="permission")
+
+
+class QuayRobotV1(ConfiguredBaseModel):
+    name: str = Field(..., alias="name")
+    description: Optional[str] = Field(..., alias="description")
+    quay_org: Optional[QuayOrgV1] = Field(..., alias="quay_org")
+    teams: Optional[list[str]] = Field(..., alias="teams")
+    repositories: Optional[list[QuayRepositoryV1]] = Field(..., alias="repositories")
+
+
+class QuayRobotAccountsQueryData(ConfiguredBaseModel):
+    robot_accounts: Optional[list[QuayRobotV1]] = Field(..., alias="robot_accounts")
+
+
+def query(query_func: Callable, **kwargs: Any) -> QuayRobotAccountsQueryData:
+    """
+    This is a convenience function which queries and parses the data into
+    concrete types. It should be compatible with most GQL clients.
+    You do not have to use it to consume the generated data classes.
+    Alternatively, you can also mime and alternate the behavior
+    of this function in the caller.
+
+    Parameters:
+        query_func (Callable): Function which queries your GQL Server
+        kwargs: optional arguments that will be passed to the query function
+
+    Returns:
+        QuayRobotAccountsQueryData: queried data parsed into generated classes
+    """
+    raw_data: dict[Any, Any] = query_func(DEFINITION, **kwargs)
+    return QuayRobotAccountsQueryData(**raw_data)

reconcile/quay_base.py

@@ -34,15 +34,15 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
         self.cleanup()
 
 
-def get_quay_api_store() -> dict[OrgKey, OrgInfo]:
+def get_quay_api_store() -> QuayApiStore:
     """
     Returns a dictionary with a key for each Quay organization
     managed in app-interface.
     """
     quay_orgs = queries.get_quay_orgs()
     settings = queries.get_app_interface_settings()
     secret_reader = SecretReader(settings=settings)
-    store = {}
+    store = QuayApiStore()
     for org_data in quay_orgs:
         instance_name = org_data["instance"]["name"]
         org_name = org_data["name"]