edraft
545540d05d
All checks were successful
Test before pr merge / test-lint (pull_request) Successful in 7s
148 lines
4.8 KiB
Python
148 lines
4.8 KiB
Python
from starlette.responses import JSONResponse
|
|
|
|
from cpl.dependency.event_bus import EventBusABC
|
|
from cpl.graphql.event_bus.memory import InMemoryEventBus
|
|
from queries.cities import CityGraphType, CityFilter, CitySort
|
|
from queries.hello import UserGraphType # , UserFilter, UserSort, UserGraphType
|
|
from queries.user import UserFilter, UserSort
|
|
from cpl.api.api_module import ApiModule
|
|
from cpl.application.application_builder import ApplicationBuilder
|
|
from cpl.auth.schema import User, Role
|
|
from cpl.core.configuration import Configuration
|
|
from cpl.core.console import Console
|
|
from cpl.core.environment import Environment
|
|
from cpl.core.utils.cache import Cache
|
|
from cpl.database.mysql.mysql_module import MySQLModule
|
|
from cpl.graphql.application.graphql_app import GraphQLApp
|
|
from cpl.graphql.auth.graphql_auth_module import GraphQLAuthModule
|
|
from cpl.graphql.graphql_module import GraphQLModule
|
|
from model.author_dao import AuthorDao
|
|
from model.author_query import AuthorGraphType, AuthorFilter, AuthorSort
|
|
from model.post_dao import PostDao
|
|
from model.post_query import PostFilter, PostSort, PostGraphType, PostMutation, PostSubscription
|
|
from permissions import PostPermissions
|
|
from queries.hello import HelloQuery
|
|
from scoped_service import ScopedService
|
|
from service import PingService
|
|
from test_data_seeder import TestDataSeeder
|
|
|
|
|
|
def main():
|
|
builder = ApplicationBuilder[GraphQLApp](GraphQLApp)
|
|
|
|
Configuration.add_json_file(f"appsettings.json")
|
|
Configuration.add_json_file(f"appsettings.{Environment.get_environment()}.json")
|
|
Configuration.add_json_file(f"appsettings.{Environment.get_host_name()}.json", optional=True)
|
|
|
|
# builder.services.add_logging()
|
|
(
|
|
builder.services.add_structured_logging()
|
|
.add_transient(PingService)
|
|
.add_module(MySQLModule)
|
|
.add_module(ApiModule)
|
|
.add_module(GraphQLModule)
|
|
.add_module(GraphQLAuthModule)
|
|
.add_scoped(ScopedService)
|
|
.add_singleton(EventBusABC, InMemoryEventBus)
|
|
.add_cache(User)
|
|
.add_cache(Role)
|
|
.add_transient(CityGraphType)
|
|
.add_transient(CityFilter)
|
|
.add_transient(CitySort)
|
|
.add_transient(UserGraphType)
|
|
.add_transient(UserFilter)
|
|
.add_transient(UserSort)
|
|
# .add_transient(UserGraphType)
|
|
# .add_transient(UserFilter)
|
|
# .add_transient(UserSort)
|
|
.add_transient(HelloQuery)
|
|
# test data
|
|
.add_singleton(TestDataSeeder)
|
|
# authors
|
|
.add_transient(AuthorDao)
|
|
.add_transient(AuthorGraphType)
|
|
.add_transient(AuthorFilter)
|
|
.add_transient(AuthorSort)
|
|
# posts
|
|
.add_transient(PostDao)
|
|
.add_transient(PostGraphType)
|
|
.add_transient(PostFilter)
|
|
.add_transient(PostSort)
|
|
.add_transient(PostMutation)
|
|
.add_transient(PostSubscription)
|
|
)
|
|
|
|
app = builder.build()
|
|
app.with_logging()
|
|
app.with_migrations("./scripts")
|
|
|
|
app.with_authentication()
|
|
app.with_authorization()
|
|
|
|
app.with_route(
|
|
path="/route1",
|
|
fn=lambda r: JSONResponse("route1"),
|
|
method="GET",
|
|
# authentication=True,
|
|
# permissions=[Permissions.administrator],
|
|
)
|
|
app.with_routes_directory("routes")
|
|
|
|
schema = app.with_graphql()
|
|
schema.query.string_field("ping", resolver=lambda: "pong")
|
|
schema.query.with_query("hello", HelloQuery)
|
|
schema.query.dao_collection_field(AuthorGraphType, AuthorDao, "authors", AuthorFilter, AuthorSort)
|
|
(
|
|
schema.query.dao_collection_field(PostGraphType, PostDao, "posts", PostFilter, PostSort)
|
|
# .with_require_any_permission(PostPermissions.read)
|
|
.with_public()
|
|
)
|
|
|
|
schema.mutation.with_mutation("post", PostMutation).with_public()
|
|
|
|
schema.subscription.with_subscription(PostSubscription)
|
|
|
|
app.with_auth_root_queries(True)
|
|
app.with_auth_root_mutations(True)
|
|
|
|
app.with_playground()
|
|
app.with_graphiql()
|
|
|
|
app.with_permissions(PostPermissions)
|
|
|
|
provider = builder.service_provider
|
|
user_cache = provider.get_service(Cache[User])
|
|
role_cache = provider.get_service(Cache[Role])
|
|
|
|
if role_cache == user_cache:
|
|
raise Exception("Cache service is not working")
|
|
|
|
s1 = provider.get_service(ScopedService)
|
|
s2 = provider.get_service(ScopedService)
|
|
|
|
if s1.name == s2.name:
|
|
raise Exception("Scoped service is not working")
|
|
|
|
with provider.create_scope() as scope:
|
|
s3 = scope.get_service(ScopedService)
|
|
s4 = scope.get_service(ScopedService)
|
|
|
|
if s3.name != s4.name:
|
|
raise Exception("Scoped service is not working")
|
|
|
|
if s1.name == s3.name:
|
|
raise Exception("Scoped service is not working")
|
|
|
|
Console.write_line(
|
|
s1.name,
|
|
s2.name,
|
|
s3.name,
|
|
s4.name,
|
|
)
|
|
|
|
app.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|