Query Params support for query methods

[i1Fury]

Add query params with safety measures against sqli into full query methods(such as the execute method).

Look at how PyMySQL got this done, it will need to be modified a bit because sdb may need some type declarations.

[dentropy]

Out of curiosity do you know of any good sources you personally use for dealing with sqli

[i1Fury]

I wrote a quick fix in my personal mini surrealdb library, that just uses json.dumps()

    def escape(self, obj: Any) -> str:
        """Escape an object to be used in a query.

        Parameters
        ----------
        obj: :class:`Any`
            The object to escape.

        Returns
        -------
        :class:`str`
            The escaped object.
        """
        return dumps(obj, default=str)

    async def format_query(self, query: str, *positional_params, **absolute_params) -> str:
        """Format a query without executing it.

        Parameters
        ----------
        query: :class:`str`
            The query to execute.
        *positional_params: :class:`Any`
            The positional parameters to use for the query.
            Example: ``SELECT * FROM table WHERE id = {}``
        **absolute_params: :class:`Any`
            The absolute parameters to use for the query.
            Example: ``SELECT * FROM table WHERE id = {id}``

        Returns
        -------
        :class:`str`
            The formatted query.
        """
        if query.count("{}") != len(positional_params):
            raise ParamsError(f"Expected {query.count('{}')} positional parameters, got {len(positional_params)}.\nQuery: {query}\nPositional Parameters: {positional_params}")

        for param in absolute_params:
            if f"{{{param}}}" not in query:
                raise ParamsError(f"Expected an absolute parameter named '{param}', got none.\nQuery: {query}\nAbsolute Parameters: {absolute_params}")

        try:
            query = query.format(
                *map(self.escape, positional_params),
                **{k: self.escape(v) for k, v in absolute_params.items()}   
            )
        except (KeyError, IndexError):
            raise ParamsError(f"Invalid parameters.\nQuery: {query}\nPositional: {positional_params}\nAbsolute: {absolute_params}")
        else:
            return query

    async def execute(self, query: str, *positional_params, **absolute_params) -> List[Dict[str, Any]]:
        """Execute a query against the SurrealDB server.

        Parameters
        ----------
        query: :class:`str`
            The query to execute.
        *positional_params: :class:`Any`
            The positional parameters to use for the query.
            Example: ``SELECT * FROM table WHERE id = {}``
        **absolute_params: :class:`Any`
            The absolute parameters to use for the query.
            Example: ``SELECT * FROM table WHERE id = {id}``

        Returns
        -------
        :class:`List[Dict[str, Any]]`
            The results of the query.
        """
        query = await self.format_query(query, *positional_params, **absolute_params)

        try:
            return await super().execute(query)
        except SurrealException as e:
            raise SurrealException(f'\nQuery: {query}\n{"".join(e.args)}') from e
        except ConnectTimeout:
            raise DBConnectionError(f"Failed to connect to the database.\nQuery: {query}")

[AlexFrid]

Thanks for the suggestions ❤️ Query parameters have been implemented in the above PR. Its now functional in the default websocket client, also using json.dumps(). It has also been implemented in the HTTP client as well, but returns an empty array currently as the HTTP query parameter functionality will become active with the beta9 release soon.

Since it has been implemented, I will therefore close this issue.