edraft
545540d05d
All checks were successful
Test before pr merge / test-lint (pull_request) Successful in 7s
149 lines
4.7 KiB
Python
149 lines
4.7 KiB
Python
from cpl.dependency.event_bus import EventBusABC
|
|
from cpl.graphql.query_context import QueryContext
|
|
from cpl.graphql.schema.db_model_graph_type import DbModelGraphType
|
|
from cpl.graphql.schema.filter.db_model_filter import DbModelFilter
|
|
from cpl.graphql.schema.input import Input
|
|
from cpl.graphql.schema.mutation import Mutation
|
|
from cpl.graphql.schema.sort.sort import Sort
|
|
from cpl.graphql.schema.sort.sort_order import SortOrder
|
|
from cpl.graphql.schema.subscription import Subscription
|
|
from model.author_dao import AuthorDao
|
|
from model.author_query import AuthorGraphType, AuthorFilter
|
|
from model.post import Post
|
|
from model.post_dao import PostDao
|
|
|
|
|
|
class PostFilter(DbModelFilter[Post]):
|
|
def __init__(self):
|
|
DbModelFilter.__init__(self, public=True)
|
|
self.int_field("id")
|
|
self.filter_field("author", AuthorFilter)
|
|
self.string_field("title")
|
|
self.string_field("content")
|
|
|
|
|
|
class PostSort(Sort[Post]):
|
|
def __init__(self):
|
|
Sort.__init__(self)
|
|
self.field("id", SortOrder)
|
|
self.field("title", SortOrder)
|
|
self.field("content", SortOrder)
|
|
|
|
|
|
class PostGraphType(DbModelGraphType[Post]):
|
|
|
|
def __init__(self, authors: AuthorDao):
|
|
DbModelGraphType.__init__(self, public=True)
|
|
|
|
self.int_field(
|
|
"id",
|
|
resolver=lambda root: root.id,
|
|
).with_optional().with_public(True)
|
|
|
|
async def _a(root: Post):
|
|
return await authors.get_by_id(root.author_id)
|
|
|
|
def r_name(ctx: QueryContext):
|
|
return ctx.user.username == "admin"
|
|
|
|
self.object_field("author", AuthorGraphType, resolver=_a).with_public(True) # .with_require_any([], [r_name]))
|
|
self.string_field(
|
|
"title",
|
|
resolver=lambda root: root.title,
|
|
).with_public(True)
|
|
self.string_field(
|
|
"content",
|
|
resolver=lambda root: root.content,
|
|
).with_public(True)
|
|
|
|
|
|
class PostCreateInput(Input[Post]):
|
|
title: str
|
|
content: str
|
|
author_id: int
|
|
|
|
def __init__(self):
|
|
Input.__init__(self)
|
|
self.string_field("title").with_required()
|
|
self.string_field("content").with_required()
|
|
self.int_field("author_id").with_required()
|
|
|
|
|
|
class PostUpdateInput(Input[Post]):
|
|
title: str
|
|
content: str
|
|
author_id: int
|
|
|
|
def __init__(self):
|
|
Input.__init__(self)
|
|
self.int_field("id").with_required()
|
|
self.string_field("title").with_required(False)
|
|
self.string_field("content").with_required(False)
|
|
|
|
|
|
class PostSubscription(Subscription):
|
|
def __init__(self, bus: EventBusABC):
|
|
Subscription.__init__(self)
|
|
self._bus = bus
|
|
|
|
def selector(event: Post, info) -> bool:
|
|
return event.id == 101
|
|
|
|
self.subscription_field("postChange", PostGraphType, selector).with_public()
|
|
|
|
|
|
class PostMutation(Mutation):
|
|
|
|
def __init__(self, posts: PostDao, authors: AuthorDao, bus: EventBusABC):
|
|
Mutation.__init__(self)
|
|
|
|
self._posts = posts
|
|
self._authors = authors
|
|
self._bus = bus
|
|
|
|
self.field("create", int, resolver=self.create_post).with_public().with_required().with_argument(
|
|
"input",
|
|
PostCreateInput,
|
|
).with_required()
|
|
self.field("update", bool, resolver=self.update_post).with_public().with_required().with_argument(
|
|
"input",
|
|
PostUpdateInput,
|
|
).with_required()
|
|
self.field("delete", bool, resolver=self.delete_post).with_public().with_required().with_argument(
|
|
"id",
|
|
int,
|
|
).with_required()
|
|
self.field("restore", bool, resolver=self.restore_post).with_public().with_required().with_argument(
|
|
"id",
|
|
int,
|
|
).with_required()
|
|
|
|
async def create_post(self, input: PostCreateInput) -> int:
|
|
return await self._posts.create(Post(0, input.author_id, input.title, input.content))
|
|
|
|
async def update_post(self, input: PostUpdateInput) -> bool:
|
|
post = await self._posts.get_by_id(input.id)
|
|
if post is None:
|
|
return False
|
|
|
|
post.title = input.title if input.title is not None else post.title
|
|
post.content = input.content if input.content is not None else post.content
|
|
|
|
await self._posts.update(post)
|
|
await self._bus.publish("postChange", post)
|
|
return True
|
|
|
|
async def delete_post(self, id: int) -> bool:
|
|
post = await self._posts.get_by_id(id)
|
|
if post is None:
|
|
return False
|
|
await self._posts.delete(post)
|
|
return True
|
|
|
|
async def restore_post(self, id: int) -> bool:
|
|
post = await self._posts.get_by_id(id)
|
|
if post is None:
|
|
return False
|
|
await self._posts.restore(post)
|
|
return True
|