mirror of
https://github.com/rjNemo/graphql_python_template
synced 2026-06-06 02:26:47 +00:00
create user module
This commit is contained in:
parent
50351e8b36
commit
33d63ceabf
21 changed files with 157 additions and 40 deletions
|
|
@ -4,7 +4,7 @@ from starlette.graphql import GraphQLApp
|
|||
from starlette.middleware.cors import CORSMiddleware
|
||||
|
||||
from app.schema.mutations.mutations import Mutations
|
||||
from app.schema.queries.todo import TodoQuery
|
||||
from app.schema.queries.queries import Query
|
||||
|
||||
origins = ["*"]
|
||||
methods = ["*"]
|
||||
|
|
@ -21,7 +21,7 @@ def create_app() -> FastAPI:
|
|||
allow_headers=headers,
|
||||
)
|
||||
app.add_route(
|
||||
"/", GraphQLApp(schema=graphene.Schema(query=TodoQuery, mutation=Mutations))
|
||||
"/", GraphQLApp(schema=graphene.Schema(query=Query, mutation=Mutations))
|
||||
)
|
||||
|
||||
return app
|
||||
|
|
|
|||
8
app/models/user.py
Normal file
8
app/models/user.py
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
from uuid import uuid4
|
||||
|
||||
|
||||
class User:
|
||||
def __init__(self, user_id: str = None, username: str = "X"):
|
||||
self.user_id = user_id or uuid4()
|
||||
self.username = username
|
||||
self.tasks = []
|
||||
40
app/repositories/users.py
Normal file
40
app/repositories/users.py
Normal file
|
|
@ -0,0 +1,40 @@
|
|||
from typing import List
|
||||
|
||||
from app.models.user import User
|
||||
|
||||
user_list: List[User] = [User("id", "Jane Doe"), User(username="John Doe")]
|
||||
|
||||
|
||||
def get_all_users() -> List[User]:
|
||||
return user_list
|
||||
|
||||
|
||||
def get_user_by_id(user_id: str) -> User:
|
||||
return [user for user in user_list if user.user_id == user_id][0]
|
||||
|
||||
|
||||
def add_user(username: str) -> User:
|
||||
user = User(username=username)
|
||||
user_list.append(user)
|
||||
return user
|
||||
|
||||
|
||||
def edit_user(user_id: str, data) -> User:
|
||||
user = get_user_by_id(user_id)
|
||||
|
||||
if title := data.get("title"):
|
||||
user.title = title
|
||||
if is_done := data.get("is_done"):
|
||||
user.is_done = is_done
|
||||
|
||||
return user
|
||||
|
||||
|
||||
def user_exists(user_id: str) -> bool:
|
||||
return any([user.user_id == user_id for user in user_list])
|
||||
|
||||
|
||||
def remove_user(user_id: str) -> User:
|
||||
user = get_user_by_id(user_id)
|
||||
index = user_list.index(user)
|
||||
return user_list.pop(index)
|
||||
|
|
@ -1,7 +1,7 @@
|
|||
import graphene
|
||||
|
||||
from app.schema.types.todo import TodoType
|
||||
from app.usecases.close_todo import close_todo
|
||||
from app.usecases.todo.close_todo import close_todo
|
||||
|
||||
|
||||
class CloseTodo(graphene.Mutation):
|
||||
|
|
|
|||
28
app/schema/queries/queries.py
Normal file
28
app/schema/queries/queries.py
Normal file
|
|
@ -0,0 +1,28 @@
|
|||
import graphene
|
||||
|
||||
from app.schema.types.todo import TodoListResponseField, TodoResponseField
|
||||
from .todo import resolve_get_todo, resolve_list_todos
|
||||
from .user import resolve_list_users, resolve_get_user
|
||||
from ..types.user import UserListResponseField, UserResponseField
|
||||
|
||||
|
||||
class Query(graphene.ObjectType):
|
||||
"""Schema Queries"""
|
||||
|
||||
"""Task Queries"""
|
||||
list_todos = graphene.Field(TodoListResponseField, resolver=resolve_list_todos)
|
||||
|
||||
get_todo = graphene.Field(
|
||||
TodoResponseField,
|
||||
todo_id=graphene.String(required=True),
|
||||
resolver=resolve_get_todo,
|
||||
)
|
||||
|
||||
"""User Queries"""
|
||||
list_users = graphene.Field(UserListResponseField, resolver=resolve_list_users)
|
||||
|
||||
get_user = graphene.Field(
|
||||
UserResponseField,
|
||||
user_id=graphene.String(required=True),
|
||||
resolver=resolve_get_user,
|
||||
)
|
||||
|
|
@ -1,15 +1,10 @@
|
|||
import graphene
|
||||
|
||||
from app.schema.types.todo import TodoResponseField, TodoListResponseField
|
||||
from app.usecases import read_all_todos, read_todo_by_id
|
||||
|
||||
|
||||
class TodoQuery(graphene.ObjectType):
|
||||
"""
|
||||
Defines the query and how to interact with
|
||||
"""
|
||||
|
||||
list_todos = graphene.Field(TodoListResponseField)
|
||||
from app.schema.types.todo import TodoResponseField, TodoListResponseField
|
||||
from app.usecases import read_all_todos, read_todo_by_id
|
||||
|
||||
|
||||
def resolve_list_todos(self, info) -> TodoListResponseField:
|
||||
try:
|
||||
|
|
@ -25,7 +20,6 @@ class TodoQuery(graphene.ObjectType):
|
|||
todos=todos, is_success=is_success, error_message=error_message
|
||||
)
|
||||
|
||||
get_todo = graphene.Field(TodoResponseField, todo_id=graphene.String(required=True))
|
||||
|
||||
def resolve_get_todo(self, info, todo_id: str) -> TodoResponseField:
|
||||
todo, is_success = read_todo_by_id(todo_id)
|
||||
|
|
|
|||
13
app/schema/queries/user.py
Normal file
13
app/schema/queries/user.py
Normal file
|
|
@ -0,0 +1,13 @@
|
|||
from app.schema.types.user import UserListResponseField, UserResponseField
|
||||
from app.usecases.user.read_all_users import read_all_users
|
||||
from app.usecases.user.read_user_by_id import read_user_by_id
|
||||
|
||||
|
||||
def resolve_list_users(self, info):
|
||||
users = read_all_users()
|
||||
return UserListResponseField(users=users)
|
||||
|
||||
|
||||
def resolve_get_user(self, info, user_id: str):
|
||||
user = read_user_by_id(user_id)
|
||||
return UserResponseField(user=user)
|
||||
6
app/schema/types/response.py
Normal file
6
app/schema/types/response.py
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
import graphene
|
||||
|
||||
|
||||
class ResponseField(graphene.ObjectType):
|
||||
is_success = graphene.Boolean(default_value=True)
|
||||
error_message = graphene.String()
|
||||
|
|
@ -1,5 +1,7 @@
|
|||
import graphene
|
||||
|
||||
from app.schema.types.response import ResponseField
|
||||
|
||||
|
||||
class TodoType(graphene.ObjectType):
|
||||
"""
|
||||
|
|
@ -21,11 +23,6 @@ class TodoInputType(graphene.InputObjectType):
|
|||
is_done = graphene.Boolean(default_value=False)
|
||||
|
||||
|
||||
class ResponseField(graphene.ObjectType):
|
||||
is_success = graphene.Boolean(default_value=True)
|
||||
error_message = graphene.String()
|
||||
|
||||
|
||||
class TodoResponseField(ResponseField):
|
||||
todo = graphene.Field(TodoType)
|
||||
|
||||
|
|
|
|||
16
app/schema/types/user.py
Normal file
16
app/schema/types/user.py
Normal file
|
|
@ -0,0 +1,16 @@
|
|||
import graphene
|
||||
|
||||
from app.schema.types.response import ResponseField
|
||||
|
||||
|
||||
class UserType(graphene.ObjectType):
|
||||
user_id = graphene.String()
|
||||
username = graphene.String()
|
||||
|
||||
|
||||
class UserListResponseField(ResponseField):
|
||||
users = graphene.List(UserType)
|
||||
|
||||
|
||||
class UserResponseField(ResponseField):
|
||||
user = graphene.Field(UserType)
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
from .create_todo import create_todo
|
||||
from .delete_todo import delete_todo
|
||||
from .read_all_todos import read_all_todos
|
||||
from .read_todo_by_id import read_todo_by_id
|
||||
from .update_todo import update_todo
|
||||
from app.usecases.todo.close_todo import close_todo
|
||||
from app.usecases.todo.create_todo import create_todo
|
||||
from app.usecases.todo.delete_todo import delete_todo
|
||||
from app.usecases.todo.read_all_todos import read_all_todos
|
||||
from app.usecases.todo.read_todo_by_id import read_todo_by_id
|
||||
from app.usecases.todo.update_todo import update_todo
|
||||
|
|
|
|||
0
app/usecases/todo/__init__.py
Normal file
0
app/usecases/todo/__init__.py
Normal file
0
app/usecases/user/__init__.py
Normal file
0
app/usecases/user/__init__.py
Normal file
8
app/usecases/user/read_all_users.py
Normal file
8
app/usecases/user/read_all_users.py
Normal file
|
|
@ -0,0 +1,8 @@
|
|||
from typing import List
|
||||
|
||||
from app.models.user import User
|
||||
from app.repositories.users import get_all_users
|
||||
|
||||
|
||||
def read_all_users() -> List[User]:
|
||||
return get_all_users()
|
||||
6
app/usecases/user/read_user_by_id.py
Normal file
6
app/usecases/user/read_user_by_id.py
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
from app.models.user import User
|
||||
from app.repositories.users import get_user_by_id
|
||||
|
||||
|
||||
def read_user_by_id(user_id: str) -> User:
|
||||
return get_user_by_id(user_id)
|
||||
Loading…
Reference in a new issue