Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

add typed queries for schemas #4716

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 20 additions & 0 deletions reconcile/gql_definitions/common/schemas.gql
Original file line number Diff line number Diff line change
@@ -0,0 +1,20 @@
# qenerate: plugin=pydantic_v1

query Schemas {
schemas: __schema {
types {
name
fields {
name
type {
kind
name
ofType {
kind
name
}
}
}
}
}
}
116 changes: 116 additions & 0 deletions reconcile/gql_definitions/common/schemas.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,116 @@
"""
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 Schemas {
schemas: __schema {
types {
name
fields {
name
type {
kind
name
ofType {
kind
name
}
}
}
}
}
}
"""


class ConfiguredBaseModel(BaseModel):
class Config:
smart_union=True
extra=Extra.forbid


class _TypeKind(Enum):
SCALAR = TypeKind.SCALAR
OBJECT = TypeKind.OBJECT
INTERFACE = TypeKind.INTERFACE
UNION = TypeKind.UNION
ENUM = TypeKind.ENUM
INPUT_OBJECT = TypeKind.INPUT_OBJECT
LIST = TypeKind.LIST
NON_NULL = TypeKind.NON_NULL


class _Field__Type__Type__TypeKind(Enum):
SCALAR = TypeKind.SCALAR
OBJECT = TypeKind.OBJECT
INTERFACE = TypeKind.INTERFACE
UNION = TypeKind.UNION
ENUM = TypeKind.ENUM
INPUT_OBJECT = TypeKind.INPUT_OBJECT
LIST = TypeKind.LIST
NON_NULL = TypeKind.NON_NULL


class _Field__Type__Type(ConfiguredBaseModel):
kind: _Field__Type__Type__TypeKind = Field(alias="kind")
name: Optional[str] = Field(..., alias="name")


class _Field__Type(ConfiguredBaseModel):
kind: _TypeKind = Field(alias="kind")
name: Optional[str] = Field(..., alias="name")
of_type: Optional[_Field__Type__Type] = Field(..., alias="ofType")


class _Field(ConfiguredBaseModel):
name: str = Field(..., alias="name")
q_type: _Field__Type = Field(..., alias="type")


class _Type(ConfiguredBaseModel):
name: Optional[str] = Field(..., alias="name")
fields: Optional[list[_Field]] = Field(..., alias="fields")


class _Schema(ConfiguredBaseModel):
types: list[_Type] = Field(..., alias="types")


class SchemasQueryData(ConfiguredBaseModel):
schemas: _Schema = Field(..., alias="schemas")


def query(query_func: Callable, **kwargs: Any) -> SchemasQueryData:
"""
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:
SchemasQueryData: queried data parsed into generated classes
"""
raw_data: dict[Any, Any] = query_func(DEFINITION, **kwargs)
return SchemasQueryData(**raw_data)
7 changes: 7 additions & 0 deletions reconcile/typed_queries/schemas.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,7 @@
from reconcile.gql_definitions.common.schemas import _Schema, query
from reconcile.utils import gql


def get_schemas() -> list[_Schema]:
data = query(gql.get_api().query)
return list(data.schemas or [])