Skip to content

Setup

The Setup class manages database engines and permission checks.

Setup

A configuration class responsible for setting up and managing database engines, executing queries, and facilitating permission checks for GraphQL operations.

Attributes:

Name Type Description
engine Dict[str, Engine]

A dictionary of SQLAlchemy engine instances indexed by name.
get_permission callable

A function to determine if a request is permitted based on custom business logic.
async_engine bool

A flag indicating if the engine configuration supports asynchronous operations.
classes Dict[str, GraftModel]

A registry of model classes that might be involved in GraphQL queries or mutations.
Source code in graftql/core/setup.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class Setup:
    """
    A configuration class responsible for setting up and managing database engines, executing queries,
    and facilitating permission checks for GraphQL operations.

    Attributes:
        engine (Dict[str, Engine]): A dictionary of SQLAlchemy engine instances indexed by name.
        get_permission (callable): A function to determine if a request is permitted based on
            custom business logic.
        async_engine (bool): A flag indicating if the engine configuration supports asynchronous
            operations.
        classes (Dict[str, 'GraftModel']): A registry of model classes that might be involved in GraphQL
            queries or mutations.
    """

    engine: dict[str, Engine] = None
    permission_getter: Callable
    async_engine = False
    classes: dict[str, "GraftModel"] = {}

    @classmethod
    async def execute_query(cls, query, engine) -> list:
        """
        Executes a given SQL query using the specified database engine, either asynchronously or synchronously
        based on the engine configuration.

        Args:
            query (Select): The SQL query to be executed.
            engine (str): The key of the engine in the 'engine' attribute to be used for the query.

        Returns:
            list: The result of the query execution, typically a list of database records.
        """
        if cls.async_engine:
            async_session = sessionmaker(
                cls.engine[engine], class_=AsyncSession, expire_on_commit=False
            )
            async with async_session() as session:
                r = await session.execute(query)
                return r.scalars().all()
        else:
            with Session(cls.engine[engine]) as session:
                r = session.exec(query).all()
                return r

    @classmethod
    def setup(
        cls,
        engine: dict[str, Engine] | Engine,
        permission_getter=None,
        query_filter=None,
    ):
        """
        Configures the database engines and sets default functions for permission checks and query filtering.

        Args:
            engine (Dict[str, Engine] | Engine): A dictionary of engines or a single engine to be used.
            permission_getter (callable): A function to determine if a request is permitted.
            query_filter (callable): A function to filter queries based on specific conditions.
        """

        if isinstance(engine, dict):
            cls.engine = engine
        else:
            cls.engine = {"default": engine}
        if engine and "async" in cls.engine["default"].__module__:
            cls.async_engine = True
        if query_filter:
            cls.query_filter = query_filter
        else:

            def query_filter_default(cls, info):
                return True

            cls.query_filter = query_filter_default
        if permission_getter:
            cls.permission_getter = permission_getter
        else:

            async def permission_getter(module_class, info, type):
                return True

            cls.permission_getter = permission_getter

    @classmethod
    async def has_permission(cls, module: "GraftModel", context: dict, request_type: str) -> bool:
        """
        Determines if a user has permission to execute a GraphQL query or mutation based on the
        provided context and request type.

        Args:
            module ('GraftModel'): The model class for which permissions are being checked.
            context (dict): The context of the GraphQL request.
            request_type (str): The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

        Returns:
            A boolean indicating if the user has permission to execute the request.
        """
        permission = await module.permission_getter(context, request_type)
        if isinstance(permission, bool):
            return permission
        else:
            permission = await cls.permission_getter(module, context, request_type)
        return permission

    @classmethod
    def get_auth(cls, module: "GraftModel", request_type: str) -> BasePermission:
        """
        Creates a custom Strawberry GraphQL permission class that performs authentication and authorization checks.

        Args:
            module ('GraftModel'): The model class for which permissions are being checked.
            request_type (str): The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

        Returns:
            An instance of a Strawberry permission class that can be used to control access to GraphQL operations.
        """

        class IsAuthenticated(BasePermission):
            async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool:
                context_with_args = dict(info.context)
                context_with_args["__resolver_args__"] = kwargs
                # persist resolver args for downstream resolvers
                with contextlib.suppress(Exception):
                    info.context["__resolver_args__"] = kwargs
                if not await cls.has_permission(module, context_with_args, request_type):
                    info.context["response"].status_code = 403
                    if "errors" not in info.context["request"].scope:
                        info.context["request"].scope["errors"] = [module]
                    elif module not in info.context["request"].scope["errors"]:
                        info.context["request"].scope["errors"].append(module)
                return True

        return IsAuthenticated

execute_query(query, engine) async classmethod

Executes a given SQL query using the specified database engine, either asynchronously or synchronously based on the engine configuration.

Parameters:

Name Type Description Default
query Select

The SQL query to be executed.

 required
engine str

The key of the engine in the 'engine' attribute to be used for the query.

 required

Returns:

Name Type Description
list list

The result of the query execution, typically a list of database records.
Source code in graftql/core/setup.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
@classmethod
async def execute_query(cls, query, engine) -> list:
    """
    Executes a given SQL query using the specified database engine, either asynchronously or synchronously
    based on the engine configuration.

    Args:
        query (Select): The SQL query to be executed.
        engine (str): The key of the engine in the 'engine' attribute to be used for the query.

    Returns:
        list: The result of the query execution, typically a list of database records.
    """
    if cls.async_engine:
        async_session = sessionmaker(
            cls.engine[engine], class_=AsyncSession, expire_on_commit=False
        )
        async with async_session() as session:
            r = await session.execute(query)
            return r.scalars().all()
    else:
        with Session(cls.engine[engine]) as session:
            r = session.exec(query).all()
            return r

get_auth(module, request_type) classmethod

Creates a custom Strawberry GraphQL permission class that performs authentication and authorization checks.

Parameters:

Name Type Description Default
module GraftModel

The model class for which permissions are being checked.

 required
request_type str

The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

 required

Returns:

Type Description
BasePermission

An instance of a Strawberry permission class that can be used to control access to GraphQL operations.
Source code in graftql/core/setup.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
@classmethod
def get_auth(cls, module: "GraftModel", request_type: str) -> BasePermission:
    """
    Creates a custom Strawberry GraphQL permission class that performs authentication and authorization checks.

    Args:
        module ('GraftModel'): The model class for which permissions are being checked.
        request_type (str): The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

    Returns:
        An instance of a Strawberry permission class that can be used to control access to GraphQL operations.
    """

    class IsAuthenticated(BasePermission):
        async def has_permission(self, source, info: strawberry.Info, **kwargs) -> bool:
            context_with_args = dict(info.context)
            context_with_args["__resolver_args__"] = kwargs
            # persist resolver args for downstream resolvers
            with contextlib.suppress(Exception):
                info.context["__resolver_args__"] = kwargs
            if not await cls.has_permission(module, context_with_args, request_type):
                info.context["response"].status_code = 403
                if "errors" not in info.context["request"].scope:
                    info.context["request"].scope["errors"] = [module]
                elif module not in info.context["request"].scope["errors"]:
                    info.context["request"].scope["errors"].append(module)
            return True

    return IsAuthenticated

has_permission(module, context, request_type) async classmethod

Determines if a user has permission to execute a GraphQL query or mutation based on the provided context and request type.

Parameters:

Name Type Description Default
module GraftModel

The model class for which permissions are being checked.

 required
context dict

The context of the GraphQL request.

 required
request_type str

The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

 required

Returns:

Type Description
bool

A boolean indicating if the user has permission to execute the request.
Source code in graftql/core/setup.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@classmethod
async def has_permission(cls, module: "GraftModel", context: dict, request_type: str) -> bool:
    """
    Determines if a user has permission to execute a GraphQL query or mutation based on the
    provided context and request type.

    Args:
        module ('GraftModel'): The model class for which permissions are being checked.
        context (dict): The context of the GraphQL request.
        request_type (str): The type of request (e.g., 'query' or 'mutation') to determine the appropriate permissions.

    Returns:
        A boolean indicating if the user has permission to execute the request.
    """
    permission = await module.permission_getter(context, request_type)
    if isinstance(permission, bool):
        return permission
    else:
        permission = await cls.permission_getter(module, context, request_type)
    return permission

setup(engine, permission_getter=None, query_filter=None) classmethod

Configures the database engines and sets default functions for permission checks and query filtering.

Parameters:

Name Type Description Default
engine Dict[str, Engine] | Engine

A dictionary of engines or a single engine to be used.

 required
permission_getter callable

A function to determine if a request is permitted.

 None
query_filter callable

A function to filter queries based on specific conditions.

 None
Source code in graftql/core/setup.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
@classmethod
def setup(
    cls,
    engine: dict[str, Engine] | Engine,
    permission_getter=None,
    query_filter=None,
):
    """
    Configures the database engines and sets default functions for permission checks and query filtering.

    Args:
        engine (Dict[str, Engine] | Engine): A dictionary of engines or a single engine to be used.
        permission_getter (callable): A function to determine if a request is permitted.
        query_filter (callable): A function to filter queries based on specific conditions.
    """

    if isinstance(engine, dict):
        cls.engine = engine
    else:
        cls.engine = {"default": engine}
    if engine and "async" in cls.engine["default"].__module__:
        cls.async_engine = True
    if query_filter:
        cls.query_filter = query_filter
    else:

        def query_filter_default(cls, info):
            return True

        cls.query_filter = query_filter_default
    if permission_getter:
        cls.permission_getter = permission_getter
    else:

        async def permission_getter(module_class, info, type):
            return True

        cls.permission_getter = permission_getter