App management
Turbulette app
Base classes to manage Turbulette apps.
TurbuletteApp
Class representing a Turbulette application.
load_directives(self)
Load app GraphQL directives.
Directives class must have a class attribute name
that match the directive name in the GraphQL schema
Source code in turbulette/apps/app.py
def load_directives(self) -> None:
"""Load app GraphQL directives.
Directives class must have a class attribute ``name``
that match the directive name in the GraphQL schema
"""
if (
not self.directives
and (self.package_path / f"{self.directives_module}.py").is_file()
):
# Relative import
app_directives_module = import_module(
f".{self.directives_module}", f"{self.package_name}"
)
for _, member in getmembers(app_directives_module, isclass):
if (
issubclass(member, SchemaDirectiveVisitor)
and member is not SchemaDirectiveVisitor
):
self.directives[member.name] = member
load_graphql_ressources(self)
Load all needed resources to enable GraphQL queries.
Source code in turbulette/apps/app.py
def load_graphql_ressources(self) -> None:
"""Load all needed resources to enable GraphQL queries."""
self.load_schema()
self.load_directives()
self.load_resolvers()
self.ready = True
load_models(self)
Load app GINO models.
Source code in turbulette/apps/app.py
def load_models(self) -> None:
"""Load app GINO models."""
if (self.package_path / f"{self.models_module}.py").is_file():
_star_import(f"{self.package_name}.{self.models_module}")
load_pydantic_models(self)
Load pydantic models.
Only models subclassing GraphQLModel are loaded.
Returns:
| Type | Description |
|---|---|
Dict[str, Type[turbulette.validation.pyd_model.GraphQLModel]] |
A dict containing all loaded pydantic models for the current app |
Source code in turbulette/apps/app.py
def load_pydantic_models(self) -> Dict[str, Type[GraphQLModel]]:
"""Load pydantic models.
Only models subclassing
[GraphQLModel][turbulette.validation.pyd_model.GraphQLModel]
are loaded.
Returns:
A dict containing all loaded pydantic models for the current app
"""
models: Dict[str, Type[GraphQLModel]] = {}
if (self.package_path / f"{self.pydantic_module}.py").is_file():
module = import_module(f".{self.pydantic_module}", f"{self.package_name}")
for _, member in getmembers(module, isclass):
if (
issubclass(member, BaseModel)
and hasattr(member, "GraphQL")
and member is not BaseModel
and member is not GraphQLModel
):
models[member.GraphQL.gql_type] = member
return models
load_resolvers(self)
Load app resolvers.
This assumes that all python modules under the resolvers
package contains resolvers functions.
non-resolver functions should live outside this folder to avoid
unnecessary imports at startup.
Resolvers defined outside the resolvers package won't be loaded
Source code in turbulette/apps/app.py
def load_resolvers(self) -> None:
"""Load app resolvers.
This assumes that all python modules under the `resolvers`
package contains resolvers functions.
non-resolver functions should live outside this folder to avoid
unnecessary imports at startup.
Resolvers defined outside the `resolvers` package won't be loaded
"""
resolver_modules = [
m.as_posix()[:-3]
.replace(sep, ".")
.split(f"{self.package_name}.{self.resolvers_package}.")[1]
for m in (self.package_path / f"{self.resolvers_package}").rglob("*.py")
]
if len(resolver_modules) > 0:
for module in resolver_modules:
# Relative import
import_module(
f".{module}", f"{self.package_name}.{self.resolvers_package}"
)
load_schema(self)
Load GraphQL schema.
Only GraphQL files under the graphql folder will be loaded.
Source code in turbulette/apps/app.py
def load_schema(self) -> None:
"""Load GraphQL schema.
Only GraphQL files under the `graphql` folder will be loaded.
"""
if (self.package_path / self.schema_folder).is_dir() and self.schema is None:
type_defs = [
load_schema_from_path(
str((self.package_path / f"{self.schema_folder}").resolve())
)
]
if type_defs:
self.schema = [*type_defs]
Registry
The registry stores and manages Turbulette apps.
Registry
A class storing the Turbulette applications in use.
It mostly serve as a proxy to execute common actions on all apps plus some configuration stuff (loading settings, models etc)
ready: bool
property
writable
The registry is ready if all of its apps are ready.
get_app_by_label(self, label)
Retrieve the Turbulette app given its label.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
label |
str |
App label |
required |
Returns:
| Type | Description |
|---|---|
TurbuletteApp |
TurbuletteApp: The Turbulette application |
Source code in turbulette/apps/registry.py
def get_app_by_label(self, label: str) -> TurbuletteApp:
"""Retrieve the Turbulette app given its label.
Args:
label: App label
Returns:
TurbuletteApp: The Turbulette application
"""
try:
return self.apps[label]
except KeyError as error:
raise RegistryError(
f'App with label "{label}" cannot be found in the registry'
) from error
get_app_by_package(self, package_name)
Retrieve a Turbulette application given its package path.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
path |
|
The module path of the app (dotted path) |
required |
Returns:
| Type | Description |
|---|---|
TurbuletteApp |
TurbuletteApp: The Turbulette application |
Source code in turbulette/apps/registry.py
def get_app_by_package(self, package_name: str) -> TurbuletteApp:
"""Retrieve a Turbulette application given its package path.
Args:
path: The module path of the app (dotted path)
Returns:
TurbuletteApp: The Turbulette application
"""
try:
return self.apps[package_name.rsplit(".", maxsplit=1)[-1]]
except KeyError as error:
raise RegistryError(
f'App "{package_name}" cannot be found in the registry'
) from error
load_models(self)
Import GINO models of each app.
This make the GINO instance aware of all models across installed apps.
Source code in turbulette/apps/registry.py
def load_models(self):
"""Import GINO models of each app.
This make the GINO instance aware of all models across installed apps.
"""
for app in self.apps.values():
app.load_models()
load_settings(self)
Put Turbulette app settings together in a LazySettings object.
The LazySettings object is from simple_settings library, which accepts
multiple modules path during instantiation.
Returns:
| Type | Description |
|---|---|
LazySettings |
LazySettings: LazySettings initialized with all settings module found |
Source code in turbulette/apps/registry.py
def load_settings(self) -> LazySettings:
"""Put Turbulette app settings together in a `LazySettings` object.
The `LazySettings` object is from `simple_settings` library, which accepts
multiple modules path during instantiation.
Returns:
LazySettings: LazySettings initialized with all settings module found
"""
if not conf.settings:
all_settings = []
for app in self.apps.values():
if (app.package_path / f"{self.settings_module}.py").is_file():
all_settings.append(f"{app.package_name}.{self.settings_module}")
# Add project settings at last position to let user
# defined settings overwrite default ones
all_settings.append(self.project_settings_path)
settings = LazySettings(*all_settings)
settings._dict[ # pylint: disable=protected-access
"SIMPLE_SETTINGS"
] = conf.SIMPLE_SETTINGS
settings.strategies = (TurbuletteSettingsLoadStrategy,)
# Make settings variable availble in the `turbulette.conf` package
conf.settings = settings
return settings
return conf.settings
register(self, app)
Register an existing [TurbuletteApp][turbulette.apps.app:TurbuletteApp].
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
app |
TurbuletteApp |
The [TurbuletteApp][turbulette.apps.app:TurbuletteApp] to add |
required |
Exceptions:
| Type | Description |
|---|---|
RegistryError |
Raised if the app is already registered |
Source code in turbulette/apps/registry.py
def register(self, app: TurbuletteApp):
"""Register an existing [TurbuletteApp][turbulette.apps.app:TurbuletteApp].
Args:
app: The [TurbuletteApp][turbulette.apps.app:TurbuletteApp] to add
Raises:
RegistryError: Raised if the app is already registered
"""
if app.label in self.apps:
raise RegistryError(f'App "{app}" is already registered')
self.apps[app.label] = app
setup(self)
Load GraphQL resources, settings and create the global executable schema.
Returns:
| Type | Description |
|---|---|
GraphQLSchema |
GraphQLSchema: The aggregated schema from all apps |
Source code in turbulette/apps/registry.py
def setup(self) -> GraphQLSchema:
"""Load GraphQL resources, settings and create the global executable schema.
Returns:
GraphQLSchema: The aggregated schema from all apps
"""
settings = self.load_settings()
if settings.APOLLO_FEDERATION:
make_schema = getattr(
import_module("ariadne.contrib.federation"), "make_federated_schema"
)
else:
make_schema = getattr(import_module("ariadne"), "make_executable_schema")
schema, directives = [], {}
for app in self.apps.values():
app.load_graphql_ressources()
app.load_models()
pydantic_binder.models.update(app.load_pydantic_models())
if app.schema:
schema.extend([*app.schema])
directives.update(app.directives)
if not schema:
raise RegistryError("None of the Turbulette apps have a schema")
executable_schema = make_schema(
[*schema],
root_mutation,
root_query,
root_subscription,
base_scalars_resolvers,
snake_case_fallback_resolvers,
pydantic_binder,
directives=None if directives == {} else directives,
)
self.ready = True
self.schema = executable_schema
return executable_schema