ariadne-codegen

import asyncio import os from pathlib import Path # --- Setup: Create dummy schema and queries files --- # In a real scenario, these files would exist in your project. schema_content = ''' schema { query: Query mutation: Mutation subscription: Subscription } type Query { users(country: String): [User!]! hello: String! } type Mutation { createUser(name: String!, email: String!): User! } type Subscription { randomNumber: Int! } type User { id: ID! name: String! email: String! country: String } ''' queries_content = ''' query ListAllUsers { users { id name email country } } mutation CreateNewUser($name: String!, $email: String!) { createUser(name: $name, email: $email) { id name email } } ''' # Create a temporary directory for the quickstart quickstart_dir = Path("./ariadne_codegen_quickstart") quickstart_dir.mkdir(exist_ok=True) schema_path = quickstart_dir / "schema.graphql" queries_path = quickstart_dir / "queries.graphql" schema_path.write_text(schema_content) queries_path.write_text(queries_content) # --- Step 1: Configure ariadne-codegen in pyproject.toml --- pyproject_toml_content = f''' [tool.ariadne-codegen] schema_path = "{schema_path}" queries_path = "{queries_path}" target_package_name = "my_graphql_client" target_package_path = "." ''' pyproject_toml_path = quickstart_dir / "pyproject.toml" pyproject_toml_path.write_text(pyproject_toml_content) # --- Step 2: Run ariadne-codegen to generate the client --- print("Generating GraphQL client...") # This command needs to be run in the shell where `ariadne-codegen` is installed. # For demonstration, we simulate its effect by assuming successful generation. # In a real setup, you would run: `cd ariadne_codegen_quickstart && ariadne-codegen` # For this runnable example, we will proceed as if 'my_graphql_client' was generated. # A placeholder for actual generation: # os.system(f"cd {quickstart_dir} && ariadne-codegen") # To make this code runnable without `ariadne-codegen` actually being installed and run, # we simulate the generated client structure. In a real scenario, these files would be created. (quickstart_dir / "my_graphql_client").mkdir(exist_ok=True) (quickstart_dir / "my_graphql_client" / "__init__.py").touch() (quickstart_dir / "my_graphql_client" / "client.py").write_text( """from typing import Any, Dict, List, Optional class User: id: str name: str email: str country: Optional[str] class ListAllUsers: users: List[User] class CreateNewUser: createUser: User class AsyncBaseClient: def __init__(self, url: str, headers: Optional[Dict[str, str]] = None): self.url = url self.headers = headers if headers is not None else {} async def _execute(self, query: str, variables: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: # In a real client, this would make an actual HTTP request. # For this example, we return mock data. print(f"Executing query to {self.url} with headers: {self.headers}") print(f"Query: {query}") print(f"Variables: {variables}") if "ListAllUsers" in query: return {"data": {"users": [{"id": "1", "name": "Alice", "email": "alice@example.com", "country": "Wonderland"}]}} if "CreateNewUser" in query: return {"data": {"createUser": {"id": "2", "name": variables['name'], "email": variables['email']}}} return {"data": {}} class Client(AsyncBaseClient): async def list_all_users(self) -> ListAllUsers: query = """ query ListAllUsers { users { id name email country } } """ data = await self._execute(query) return ListAllUsers(users=[User(**user_data) for user_data in data['data']['users']]) async def create_new_user(self, name: str, email: str) -> CreateNewUser: query = """ mutation CreateNewUser($name: String!, $email: String!) { createUser(name: $name, email: $email) { id name email } } """ variables = {"name": name, "email": email} data = await self._execute(query, variables) return CreateNewUser(createUser=User(**data['data']['createUser'])) """) # Add placeholder pydantic BaseModels (actual generated models would be more complex) (quickstart_dir / "my_graphql_client" / "base_model.py").write_text( """from pydantic import BaseModel class BaseGraphQLModel(BaseModel): class Config: arbitrary_types_allowed = True """) # Ensure `my_graphql_client` is importable from the quickstart_dir import sys sys.path.insert(0, str(quickstart_dir)) from my_graphql_client.client import Client from my_graphql_client.client import ListAllUsers, CreateNewUser async def main(): # Replace with your actual GraphQL endpoint graphql_url = os.environ.get('GRAPHQL_ENDPOINT', 'https://api.example.com/graphql') # Replace with your actual authentication token if needed auth_token = os.environ.get('GRAPHQL_AUTH_TOKEN', '') headers = {} if auth_token: headers['Authorization'] = f'Bearer {auth_token}' client = Client(url=graphql_url, headers=headers) print("\n--- Listing all users ---") all_users_result: ListAllUsers = await client.list_all_users() for user in all_users_result.users: print(f"User ID: {user.id}, Name: {user.name}, Email: {user.email}, Country: {user.country}") print("\n--- Creating a new user ---") new_user_result: CreateNewUser = await client.create_new_user(name="Charlie", email="charlie@example.com") created_user = new_user_result.createUser print(f"Created User ID: {created_user.id}, Name: {created_user.name}, Email: {created_user.email}") # Clean up temporary files import shutil shutil.rmtree(quickstart_dir) print(f"\nCleaned up temporary directory: {quickstart_dir}") if __name__ == "__main__": asyncio.run(main())