Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
)
from ...validation import (
CategoryIsOpenValidator,
ThreadAuthorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
ThreadIsOpenValidator,
UserIsAuthorizedRootValidator,
threadtitlestr,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
edit_thread_title_mutation = MutationType()
@edit_thread_title_mutation.field("editThreadTitle")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_edit_thread_title(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await edit_thread_title_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("thread"):
thread = await load_thread(info.context, cleaned_data["thread"])
else:
DeleteThreadsInputModel,
)
from ...validation import (
CategoryModeratorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
ThreadsBulkValidator,
UserIsAuthorizedRootValidator,
bulkactionidslist,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
delete_threads_mutation = MutationType()
@delete_threads_mutation.field("deleteThreads")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_delete_threads(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await delete_threads_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("threads"):
# prime threads cache for bulk action
await load_threads(info.context, cleaned_data["threads"])
)
from ...validation import (
CategoryIsOpenValidator,
PostAuthorValidator,
PostCategoryValidator,
PostExistsValidator,
PostThreadValidator,
ThreadIsOpenValidator,
UserIsAuthorizedRootValidator,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
edit_post_mutation = MutationType()
@edit_post_mutation.field("editPost")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_edit_post(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await edit_post_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
post = None
thread = None
DeleteThreadReplyInputModel,
Thread,
)
from ...validation import (
CategoryModeratorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
ThreadReplyExistsValidator,
UserIsAuthorizedRootValidator,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
delete_thread_reply_mutation = MutationType()
@delete_thread_reply_mutation.field("deleteThreadReply")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_delete_thread_reply(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await delete_thread_reply_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("thread"):
thread = await load_thread(info.context, cleaned_data["thread"])
else:
from ...validation import (
CategoryExistsValidator,
CategoryIsOpenValidator,
CategoryModeratorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
ThreadsBulkValidator,
UserIsAuthorizedRootValidator,
bulkactionidslist,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
move_threads_mutation = MutationType()
@move_threads_mutation.field("moveThreads")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_move_threads(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await move_threads_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("threads"):
threads = clear_list(await load_threads(info.context, cleaned_data["threads"]))
else:
from ariadne import MutationType
from graphql import GraphQLResolveInfo
from ...auth import authenticate_user, create_user_token
from ...errors import AllFieldsAreRequiredError, InvalidCredentialsError
from ...hooks import authenticate_user_hook, create_user_token_hook
from ..decorators import error_handler
login_mutation = MutationType()
@login_mutation.field("login")
@error_handler
async def resolve_login(_, info: GraphQLResolveInfo, *, username: str, password: str):
username = str(username or "").strip()
password = str(password or "")
if not username or not password:
raise AllFieldsAreRequiredError()
user = await authenticate_user_hook.call_action(
authenticate_user, info.context, username, password
)
if not user:
RegisterUserInputModel,
User,
)
from ...users.create import create_user
from ...validation import (
EmailIsAvailableValidator,
UsernameIsAvailableValidator,
passwordstr,
usernamestr,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
register_mutation = MutationType()
@register_mutation.field("register")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_register(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await register_user_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data:
validators: Dict[str, List[AsyncValidator]] = {
"name": [UsernameIsAvailableValidator(),],
# This is our schema
type_defs = gql(
"""
type Query {
hello: String!
}
type Mutation {
add(a: Int!, b: Int!): Int!
}
"""
)
query = QueryType() # our Query type
mutation = MutationType() # our Mutation type
@query.field("hello") # Query.hello
def resolve_hello(*_):
return "Hello DjangoCon!"
@mutation.field("add") # Mutation.add
def resolve_add(*_, a: int, b: int):
return a + b
# Create an executable GraphQL schema
schema = make_executable_schema(type_defs, [query, mutation])
# Create the ASGI app
)
from ...utils.lists import clear_list
from ...validation import (
CategoryModeratorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
ThreadsBulkValidator,
UserIsAuthorizedRootValidator,
bulkactionidslist,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
close_threads_mutation = MutationType()
@close_threads_mutation.field("closeThreads")
@error_handler
@convert_kwargs_to_snake_case
async def resolve_close_threads(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await close_threads_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("threads"):
threads = clear_list(await load_threads(info.context, cleaned_data["threads"]))
else:
Thread,
)
from ...validation import (
CategoryExistsValidator,
CategoryIsOpenValidator,
CategoryModeratorValidator,
ThreadCategoryValidator,
ThreadExistsValidator,
UserIsAuthorizedRootValidator,
validate_data,
validate_model,
)
from ..errorhandler import error_handler
move_thread_mutation = MutationType()
@move_thread_mutation.field("moveThread")
@convert_kwargs_to_snake_case
@error_handler
async def resolve_move_thread(
_, info: GraphQLResolveInfo, *, input: dict # pylint: disable=redefined-builtin
):
input_model = await move_thread_input_model_hook.call_action(
create_input_model, info.context
)
cleaned_data, errors = validate_model(input_model, input)
if cleaned_data.get("thread"):
thread = await load_thread(info.context, cleaned_data["thread"])
else: