Skip to content

API Reference

Query

Source code in sirqle/query.py 
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
class Config:
    def __init__(
        self,
        env_file: str = ".db_conf",
        client: Optional[SurrealHTTP | Surreal] = None,
        url: Optional[str] = None,
        namespace: Optional[str] = None,
        database: Optional[str] = None,
        username: Optional[str] = None,
        password: Optional[str] = None,
    ) -> None:
        """Generate a Config class is used to configure the connection the database.

        It uses the `SurrealHTTP` client from the `surrealdb` library

        Args:
            env_file: the name of an env file
            client: a predefined Surreal client
            url: the URL to the database
            namespace: the namespace in the database
            database: the name of the database
            username: the username used for authentication
            password: the password used for authentication
        """
        if client:
            self.client = client
        elif os.path.isfile(env_file):
            conf = dotenv_values(env_file)
            if set(PARAMS).issubset(set(conf.keys())):
                scheme = str(urlparse(conf["URL"]).scheme)
                self.client = CLIENT[scheme](
                    **{param.lower(): conf[param] for param in PARAMS}
                )
        elif username and database and password and namespace and url:
            scheme = str(urlparse(url).scheme)
            self.client = CLIENT[scheme](
                url,
                namespace=namespace,
                database=database,
                username=username,
                password=password,
            )
        elif url:
            scheme = str(urlparse(url).scheme)
            if scheme in ["wss", "ws"]:
                self.client = CLIENT[scheme](url=url)

    async def signup(self, user: str, password: str) -> str:
        if isinstance(self.client, Surreal):
            if self.client.client_state != ConnectionState.CONNECTED:
                await self.client.connect()
        return await self.client.signup({"user": user, "pass": password})

    async def signin(self, user: str, password: str) -> str:
        if isinstance(self.client, Surreal):
            if self.client.client_state != ConnectionState.CONNECTED:
                await self.client.connect()
        return await self.client.signin({"user": user, "pass": password})

__init__(env_file='.db_conf', client=None, url=None, namespace=None, database=None, username=None, password=None)

Generate a Config class is used to configure the connection the database.

It uses the SurrealHTTP client from the surrealdb library

Parameters:

Name Type Description Default
env_file str

the name of an env file

 '.db_conf'
client Optional[SurrealHTTP | Surreal]

a predefined Surreal client

 None
url Optional[str]

the URL to the database

 None
namespace Optional[str]

the namespace in the database

 None
database Optional[str]

the name of the database

 None
username Optional[str]

the username used for authentication

 None
password Optional[str]

the password used for authentication

 None
Source code in sirqle/query.py 
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
def __init__(
    self,
    env_file: str = ".db_conf",
    client: Optional[SurrealHTTP | Surreal] = None,
    url: Optional[str] = None,
    namespace: Optional[str] = None,
    database: Optional[str] = None,
    username: Optional[str] = None,
    password: Optional[str] = None,
) -> None:
    """Generate a Config class is used to configure the connection the database.

    It uses the `SurrealHTTP` client from the `surrealdb` library

    Args:
        env_file: the name of an env file
        client: a predefined Surreal client
        url: the URL to the database
        namespace: the namespace in the database
        database: the name of the database
        username: the username used for authentication
        password: the password used for authentication
    """
    if client:
        self.client = client
    elif os.path.isfile(env_file):
        conf = dotenv_values(env_file)
        if set(PARAMS).issubset(set(conf.keys())):
            scheme = str(urlparse(conf["URL"]).scheme)
            self.client = CLIENT[scheme](
                **{param.lower(): conf[param] for param in PARAMS}
            )
    elif username and database and password and namespace and url:
        scheme = str(urlparse(url).scheme)
        self.client = CLIENT[scheme](
            url,
            namespace=namespace,
            database=database,
            username=username,
            password=password,
        )
    elif url:
        scheme = str(urlparse(url).scheme)
        if scheme in ["wss", "ws"]:
            self.client = CLIENT[scheme](url=url)

options:
  show_source: false

Config

Source code in sirqle/query.py 
 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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
class Query:
    def __init__(self, config: Config | None = None):
        """Create a `Query` class

        The Query module aims to replace the standard SurrealQL and make it more Python
            friendly. Internally it constructs a SurrealQL string from method chaining
            and sends the query to the database.

        Args:
            config: a `Config` object
        """
        self.query = ""
        self.last_query = ""
        # TODO Process config here rather than having to instantiate another
        if config:
            self.client = config.client

    def _parse_args(self, args: str | list | dict | tuple | Query, quote=True) -> str:
        if isinstance(args, str):
            if quote and ":" not in args:
                return f"'{args}'"
            return args
        elif isinstance(args, list):
            if quote:
                return "[" + ", ".join([self._parse_args(arg) for arg in args]) + "]"
            else:
                return ", ".join(args)
        elif isinstance(args, dict):
            return (
                "{"
                + "".join(
                    [
                        f"\n{key}: {self._parse_args(value)}, "
                        for key, value in args.items()
                    ]
                )
                + "\n}"
            )
        elif isinstance(args, tuple):
            return (
                "(" + ", ".join([f"{self._parse_args(value)}" for value in args]) + ")"
            )

        elif isinstance(args, Query):
            return f"({args[:-1]})"
        elif isinstance(args, int):
            return args

    def custom(self, args: str) -> Query:
        """Create a simple query in SurrealQL.


        Args:
            args: a string containing a full and correct query

        Returns:
            Query: returns the same `Query` object
        """
        self.query += args
        return self

    def select(self, args: str | List[str]) -> Query:
        """SELECT statement.
        Args:

            args: a table or a list of tables

        Returns:
            Query: returns the same `Query` object
        """
        self.query = " SELECT "
        self.query += self._parse_args(args, quote=False)
        return self

    def insert(
        self, args: str, values: Tuple[str] | List[Tuple[str]] | dict | str = ""
    ) -> Query:
        """INSERT statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += "INSERT INTO "
        self.query += self._parse_args(args, quote=False) + " "
        if values:
            if isinstance(values, (str, tuple)):
                self.query += " VALUES "
                self.query += self._parse_args(values)
            elif isinstance(values, (list, dict)):
                self.query += self._parse_args(values)
            else:
                raise NotImplementedError()
        return self

    def create(self, args: str | list | dict | Query) -> Query:
        """CREATE statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " CREATE "
        self.query += self._parse_args(args, quote=False)
        return self

    def update(self, args: str | list | dict | Query) -> Query:
        """UPDATE statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " UPDATE "
        self.query += self._parse_args(args, quote=False)
        return self

    def relate(
        self,
        node1: str,
        edge: str,
        node2: str,
        args: str | list | dict | Query | None = None,
    ) -> Query:
        """RELATE statement.

        Create a relation between two nodes.

        Args:
            node1: the source node
            edge: name of the relation
            node2: the destination node
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += f" RELATE {node1}->{edge}->{node2}"
        if args:
            self.query += " CONTENT "
            self.query += self._parse_args(args)
        return self

    def delete(self, args: str | list | dict | Query) -> Query:
        """DELETE statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " DELETE "
        self.query += self._parse_args(args, quote=False)
        return self

    def define(self, args: str | list | dict | Query) -> Query:
        """DEFINE statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " DEFINE "
        self.query += self._parse_args(args, quote=False)
        return self

    def from_(self, args: str | list | dict | Query) -> Query:
        """FROM statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " FROM "
        self.query += self._parse_args(args, quote=False)
        return self

    def where(self, args: str | list | dict | Query) -> Query:
        """WHERE statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        if args:
            self.query += " WHERE "
            self.query += self._parse_args(args, quote=False)
        else:
            warn(
                "No arguments for RETURN statement. RETURN statement not \
                added to the query."
            )
        return self

    def return_(self, args: str | list | dict | Query | None) -> Query:
        """RETURN statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        if args:
            self.query += " RETURN "
            self.query += self._parse_args(args, quote=False)
        else:
            warn(
                "No arguments for RETURN statement. RETURN statement not \
                added to the query."
            )
        return self

    def content(self, args: str | list | dict | Query) -> Query:
        """CONTENT statement.

        Args:
            args: arguments for the statement

        Returns:
            Query: returns the same `Query` object
        """
        self.query += " CONTENT "
        self.query += self._parse_args(args)
        return self

    def _end(self):
        self.query = self.query.strip()
        if self.query[-1] != ";":
            self.query += ";"

    async def _execute_query(self):
        self._end()
        if isinstance(self.client, Surreal):
            if self.client.client_state != ConnectionState.CONNECTED:
                await self.client.connect()
        res = await self.client.query(self.query)
        self.last_query = self.query
        self.query = ""
        return res

    async def execute(self):
        """Execute the query.

        Send the query to the databse and get the answert.

        """
        if self.client is None:
            raise Exception("No client provided!")
        res = await self._execute_query()
        return res

    async def use(self, ns: str, db: str) -> None:
        if isinstance(self.client, Surreal):
            await self.client.use(namespace=ns, database=db)
        elif isinstance(self.client, SurrealHTTP):
            self.client = SurrealHTTP(
                url=self.client._url,
                namespace=ns,
                database=db,
                username=self.client._username,
                password=self.client._password,
            )

    # HACK:
    def __repr__(self):
        self._end()
        return self.query

    def __getitem__(self, key):
        return self.query[key]

__init__(config=None)

Create a Query class

The Query module aims to replace the standard SurrealQL and make it more Python friendly. Internally it constructs a SurrealQL string from method chaining and sends the query to the database.

Parameters:

Name Type Description Default
config Config | None

a Config object

 None
Source code in sirqle/query.py 
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(self, config: Config | None = None):
    """Create a `Query` class

    The Query module aims to replace the standard SurrealQL and make it more Python
        friendly. Internally it constructs a SurrealQL string from method chaining
        and sends the query to the database.

    Args:
        config: a `Config` object
    """
    self.query = ""
    self.last_query = ""
    # TODO Process config here rather than having to instantiate another
    if config:
        self.client = config.client

content(args)

CONTENT statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
309
310
311
312
313
314
315
316
317
318
319
320
def content(self, args: str | list | dict | Query) -> Query:
    """CONTENT statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " CONTENT "
    self.query += self._parse_args(args)
    return self

create(args)

CREATE statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
180
181
182
183
184
185
186
187
188
189
190
191
def create(self, args: str | list | dict | Query) -> Query:
    """CREATE statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " CREATE "
    self.query += self._parse_args(args, quote=False)
    return self

custom(args)

Create a simple query in SurrealQL.

Parameters:

Name Type Description Default
args str

a string containing a full and correct query

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
131
132
133
134
135
136
137
138
139
140
141
142
def custom(self, args: str) -> Query:
    """Create a simple query in SurrealQL.


    Args:
        args: a string containing a full and correct query

    Returns:
        Query: returns the same `Query` object
    """
    self.query += args
    return self

define(args)

DEFINE statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
245
246
247
248
249
250
251
252
253
254
255
256
def define(self, args: str | list | dict | Query) -> Query:
    """DEFINE statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " DEFINE "
    self.query += self._parse_args(args, quote=False)
    return self

delete(args)

DELETE statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
232
233
234
235
236
237
238
239
240
241
242
243
def delete(self, args: str | list | dict | Query) -> Query:
    """DELETE statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " DELETE "
    self.query += self._parse_args(args, quote=False)
    return self

execute() async

Execute the query.

Send the query to the databse and get the answert.

Source code in sirqle/query.py 
337
338
339
340
341
342
343
344
345
346
async def execute(self):
    """Execute the query.

    Send the query to the databse and get the answert.

    """
    if self.client is None:
        raise Exception("No client provided!")
    res = await self._execute_query()
    return res

from_(args)

FROM statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
258
259
260
261
262
263
264
265
266
267
268
269
def from_(self, args: str | list | dict | Query) -> Query:
    """FROM statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " FROM "
    self.query += self._parse_args(args, quote=False)
    return self

insert(args, values='')

INSERT statement.

Parameters:

Name Type Description Default
args str

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
def insert(
    self, args: str, values: Tuple[str] | List[Tuple[str]] | dict | str = ""
) -> Query:
    """INSERT statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += "INSERT INTO "
    self.query += self._parse_args(args, quote=False) + " "
    if values:
        if isinstance(values, (str, tuple)):
            self.query += " VALUES "
            self.query += self._parse_args(values)
        elif isinstance(values, (list, dict)):
            self.query += self._parse_args(values)
        else:
            raise NotImplementedError()
    return self

relate(node1, edge, node2, args=None)

RELATE statement.

Create a relation between two nodes.

Parameters:

Name Type Description Default
node1 str

the source node

 required
edge str

name of the relation

 required
node2 str

the destination node

 required
args str | list | dict | Query | None

arguments for the statement

 None

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def relate(
    self,
    node1: str,
    edge: str,
    node2: str,
    args: str | list | dict | Query | None = None,
) -> Query:
    """RELATE statement.

    Create a relation between two nodes.

    Args:
        node1: the source node
        edge: name of the relation
        node2: the destination node
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += f" RELATE {node1}->{edge}->{node2}"
    if args:
        self.query += " CONTENT "
        self.query += self._parse_args(args)
    return self

return_(args)

RETURN statement.

Parameters:

Name Type Description Default
args str | list | dict | Query | None

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
def return_(self, args: str | list | dict | Query | None) -> Query:
    """RETURN statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    if args:
        self.query += " RETURN "
        self.query += self._parse_args(args, quote=False)
    else:
        warn(
            "No arguments for RETURN statement. RETURN statement not \
            added to the query."
        )
    return self

select(args)

SELECT statement.

Parameters:

Name Type Description Default
args str | List[str]

a table or a list of tables

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
144
145
146
147
148
149
150
151
152
153
154
155
def select(self, args: str | List[str]) -> Query:
    """SELECT statement.
    Args:

        args: a table or a list of tables

    Returns:
        Query: returns the same `Query` object
    """
    self.query = " SELECT "
    self.query += self._parse_args(args, quote=False)
    return self

update(args)

UPDATE statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
193
194
195
196
197
198
199
200
201
202
203
204
def update(self, args: str | list | dict | Query) -> Query:
    """UPDATE statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    self.query += " UPDATE "
    self.query += self._parse_args(args, quote=False)
    return self

where(args)

WHERE statement.

Parameters:

Name Type Description Default
args str | list | dict | Query

arguments for the statement

 required

Returns:

Name Type Description
Query Query

returns the same Query object
Source code in sirqle/query.py 
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
def where(self, args: str | list | dict | Query) -> Query:
    """WHERE statement.

    Args:
        args: arguments for the statement

    Returns:
        Query: returns the same `Query` object
    """
    if args:
        self.query += " WHERE "
        self.query += self._parse_args(args, quote=False)
    else:
        warn(
            "No arguments for RETURN statement. RETURN statement not \
            added to the query."
        )
    return self

options:
  show_source: false