-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Expand file tree
/
Copy pathexpressions.py
More file actions
10250 lines (7474 loc) · 267 KB
/
expressions.py
File metadata and controls
10250 lines (7474 loc) · 267 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
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
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
"""
## Expressions
Every AST node in SQLGlot is represented by a subclass of `Expression`.
This module contains the implementation of all supported `Expression` types. Additionally,
it exposes a number of helper functions, which are mainly used to programmatically build
SQL expressions, such as `sqlglot.expressions.select`.
----
"""
from __future__ import annotations
import datetime
import math
import numbers
import re
import sys
import textwrap
import typing as t
from collections import deque
from copy import deepcopy
from decimal import Decimal
from enum import auto
from functools import reduce
from sqlglot.errors import ErrorLevel, ParseError
from sqlglot.helper import (
AutoName,
camel_to_snake_case,
ensure_collection,
ensure_list,
seq_get,
split_num_words,
subclasses,
to_bool,
)
from sqlglot.tokens import Token, TokenError
if t.TYPE_CHECKING:
from typing_extensions import Self
from sqlglot._typing import E, Lit
from sqlglot.dialects.dialect import DialectType
Q = t.TypeVar("Q", bound="Query")
S = t.TypeVar("S", bound="SetOperation")
class _Expression(type):
def __new__(cls, clsname, bases, attrs):
klass = super().__new__(cls, clsname, bases, attrs)
# When an Expression class is created, its key is automatically set
# to be the lowercase version of the class' name.
klass.key = clsname.lower()
klass.required_args = {k for k, v in klass.arg_types.items() if v}
# This is so that docstrings are not inherited in pdoc
klass.__doc__ = klass.__doc__ or ""
return klass
SQLGLOT_META = "sqlglot.meta"
SQLGLOT_ANONYMOUS = "sqlglot.anonymous"
TABLE_PARTS = ("this", "db", "catalog")
COLUMN_PARTS = ("this", "table", "db", "catalog")
POSITION_META_KEYS = ("line", "col", "start", "end")
UNITTEST = "unittest" in sys.modules or "pytest" in sys.modules
class Expression(metaclass=_Expression):
"""
The base class for all expressions in a syntax tree. Each Expression encapsulates any necessary
context, such as its child expressions, their names (arg keys), and whether a given child expression
is optional or not.
Attributes:
key: a unique key for each class in the Expression hierarchy. This is useful for hashing
and representing expressions as strings.
arg_types: determines the arguments (child nodes) supported by an expression. It maps
arg keys to booleans that indicate whether the corresponding args are optional.
parent: a reference to the parent expression (or None, in case of root expressions).
arg_key: the arg key an expression is associated with, i.e. the name its parent expression
uses to refer to it.
index: the index of an expression if it is inside of a list argument in its parent.
comments: a list of comments that are associated with a given expression. This is used in
order to preserve comments when transpiling SQL code.
type: the `sqlglot.expressions.DataType` type of an expression. This is inferred by the
optimizer, in order to enable some transformations that require type information.
meta: a dictionary that can be used to store useful metadata for a given expression.
Example:
>>> class Foo(Expression):
... arg_types = {"this": True, "expression": False}
The above definition informs us that Foo is an Expression that requires an argument called
"this" and may also optionally receive an argument called "expression".
Args:
args: a mapping used for retrieving the arguments of an expression, given their arg keys.
"""
key = "expression"
arg_types = {"this": True}
required_args = {"this"}
__slots__ = ("args", "parent", "arg_key", "index", "comments", "_type", "_meta", "_hash")
def __init__(self, **args: t.Any):
self.args: t.Dict[str, t.Any] = args
self.parent: t.Optional[Expression] = None
self.arg_key: t.Optional[str] = None
self.index: t.Optional[int] = None
self.comments: t.Optional[t.List[str]] = None
self._type: t.Optional[DataType] = None
self._meta: t.Optional[t.Dict[str, t.Any]] = None
self._hash: t.Optional[int] = None
for arg_key, value in self.args.items():
self._set_parent(arg_key, value)
def __eq__(self, other) -> bool:
return self is other or (type(self) is type(other) and hash(self) == hash(other))
def __hash__(self) -> int:
if self._hash is None:
nodes = []
queue = deque([self])
while queue:
node = queue.popleft()
nodes.append(node)
for v in node.iter_expressions():
if v._hash is None:
queue.append(v)
for node in reversed(nodes):
hash_ = hash(node.key)
t = type(node)
if t is Literal or t is Identifier:
for k, v in sorted(node.args.items()):
if v:
hash_ = hash((hash_, k, v))
else:
for k, v in sorted(node.args.items()):
t = type(v)
if t is list:
for x in v:
if x is not None and x is not False:
hash_ = hash((hash_, k, x.lower() if type(x) is str else x))
else:
hash_ = hash((hash_, k))
elif v is not None and v is not False:
hash_ = hash((hash_, k, v.lower() if t is str else v))
node._hash = hash_
assert self._hash
return self._hash
def __reduce__(self) -> t.Tuple[t.Callable, t.Tuple[t.List[t.Dict[str, t.Any]]]]:
from sqlglot.serde import dump, load
return (load, (dump(self),))
@property
def this(self) -> t.Any:
"""
Retrieves the argument with key "this".
"""
return self.args.get("this")
@property
def expression(self) -> t.Any:
"""
Retrieves the argument with key "expression".
"""
return self.args.get("expression")
@property
def expressions(self) -> t.List[t.Any]:
"""
Retrieves the argument with key "expressions".
"""
return self.args.get("expressions") or []
def text(self, key) -> str:
"""
Returns a textual representation of the argument corresponding to "key". This can only be used
for args that are strings or leaf Expression instances, such as identifiers and literals.
"""
field = self.args.get(key)
if isinstance(field, str):
return field
if isinstance(field, (Identifier, Literal, Var)):
return field.this
if isinstance(field, (Star, Null)):
return field.name
return ""
@property
def is_string(self) -> bool:
"""
Checks whether a Literal expression is a string.
"""
return isinstance(self, Literal) and self.args["is_string"]
@property
def is_number(self) -> bool:
"""
Checks whether a Literal expression is a number.
"""
return (isinstance(self, Literal) and not self.args["is_string"]) or (
isinstance(self, Neg) and self.this.is_number
)
def to_py(self) -> t.Any:
"""
Returns a Python object equivalent of the SQL node.
"""
raise ValueError(f"{self} cannot be converted to a Python object.")
@property
def is_int(self) -> bool:
"""
Checks whether an expression is an integer.
"""
return self.is_number and isinstance(self.to_py(), int)
@property
def is_star(self) -> bool:
"""Checks whether an expression is a star."""
return isinstance(self, Star) or (isinstance(self, Column) and isinstance(self.this, Star))
@property
def alias(self) -> str:
"""
Returns the alias of the expression, or an empty string if it's not aliased.
"""
if isinstance(self.args.get("alias"), TableAlias):
return self.args["alias"].name
return self.text("alias")
@property
def alias_column_names(self) -> t.List[str]:
table_alias = self.args.get("alias")
if not table_alias:
return []
return [c.name for c in table_alias.args.get("columns") or []]
@property
def name(self) -> str:
return self.text("this")
@property
def alias_or_name(self) -> str:
return self.alias or self.name
@property
def output_name(self) -> str:
"""
Name of the output column if this expression is a selection.
If the Expression has no output name, an empty string is returned.
Example:
>>> from sqlglot import parse_one
>>> parse_one("SELECT a").expressions[0].output_name
'a'
>>> parse_one("SELECT b AS c").expressions[0].output_name
'c'
>>> parse_one("SELECT 1 + 2").expressions[0].output_name
''
"""
return ""
@property
def type(self) -> t.Optional[DataType]:
return self._type
@type.setter
def type(self, dtype: t.Optional[DataType | DataType.Type | str]) -> None:
if dtype and not isinstance(dtype, DataType):
dtype = DataType.build(dtype)
self._type = dtype # type: ignore
def is_type(self, *dtypes) -> bool:
return self.type is not None and self.type.is_type(*dtypes)
def is_leaf(self) -> bool:
return not any(isinstance(v, (Expression, list)) and v for v in self.args.values())
@property
def meta(self) -> t.Dict[str, t.Any]:
if self._meta is None:
self._meta = {}
return self._meta
def __deepcopy__(self, memo):
root = self.__class__()
stack = [(self, root)]
while stack:
node, copy = stack.pop()
if node.comments is not None:
copy.comments = deepcopy(node.comments)
if node._type is not None:
copy._type = deepcopy(node._type)
if node._meta is not None:
copy._meta = deepcopy(node._meta)
if node._hash is not None:
copy._hash = node._hash
for k, vs in node.args.items():
if hasattr(vs, "parent"):
stack.append((vs, vs.__class__()))
copy.set(k, stack[-1][-1])
elif type(vs) is list:
copy.args[k] = []
for v in vs:
if hasattr(v, "parent"):
stack.append((v, v.__class__()))
copy.append(k, stack[-1][-1])
else:
copy.append(k, v)
else:
copy.args[k] = vs
return root
def copy(self) -> Self:
"""
Returns a deep copy of the expression.
"""
return deepcopy(self)
def add_comments(self, comments: t.Optional[t.List[str]] = None, prepend: bool = False) -> None:
if self.comments is None:
self.comments = []
if comments:
for comment in comments:
_, *meta = comment.split(SQLGLOT_META)
if meta:
for kv in "".join(meta).split(","):
k, *v = kv.split("=")
value = v[0].strip() if v else True
self.meta[k.strip()] = to_bool(value)
if not prepend:
self.comments.append(comment)
if prepend:
self.comments = comments + self.comments
def pop_comments(self) -> t.List[str]:
comments = self.comments or []
self.comments = None
return comments
def append(self, arg_key: str, value: t.Any) -> None:
"""
Appends value to arg_key if it's a list or sets it as a new list.
Args:
arg_key (str): name of the list expression arg
value (Any): value to append to the list
"""
if type(self.args.get(arg_key)) is not list:
self.args[arg_key] = []
self._set_parent(arg_key, value)
values = self.args[arg_key]
if hasattr(value, "parent"):
value.index = len(values)
values.append(value)
def set(
self,
arg_key: str,
value: t.Any,
index: t.Optional[int] = None,
overwrite: bool = True,
) -> None:
"""
Sets arg_key to value.
Args:
arg_key: name of the expression arg.
value: value to set the arg to.
index: if the arg is a list, this specifies what position to add the value in it.
overwrite: assuming an index is given, this determines whether to overwrite the
list entry instead of only inserting a new value (i.e., like list.insert).
"""
expression: t.Optional[Expression] = self
while expression and expression._hash is not None:
expression._hash = None
expression = expression.parent
if index is not None:
expressions = self.args.get(arg_key) or []
if seq_get(expressions, index) is None:
return
if value is None:
expressions.pop(index)
for v in expressions[index:]:
v.index = v.index - 1
return
if isinstance(value, list):
expressions.pop(index)
expressions[index:index] = value
elif overwrite:
expressions[index] = value
else:
expressions.insert(index, value)
value = expressions
elif value is None:
self.args.pop(arg_key, None)
return
self.args[arg_key] = value
self._set_parent(arg_key, value, index)
def _set_parent(self, arg_key: str, value: t.Any, index: t.Optional[int] = None) -> None:
if hasattr(value, "parent"):
value.parent = self
value.arg_key = arg_key
value.index = index
elif type(value) is list:
for index, v in enumerate(value):
if hasattr(v, "parent"):
v.parent = self
v.arg_key = arg_key
v.index = index
@property
def depth(self) -> int:
"""
Returns the depth of this tree.
"""
if self.parent:
return self.parent.depth + 1
return 0
def iter_expressions(self, reverse: bool = False) -> t.Iterator[Expression]:
"""Yields the key and expression for all arguments, exploding list args."""
for vs in reversed(self.args.values()) if reverse else self.args.values(): # type: ignore
if type(vs) is list:
for v in reversed(vs) if reverse else vs: # type: ignore
if hasattr(v, "parent"):
yield v
elif hasattr(vs, "parent"):
yield vs
def find(self, *expression_types: t.Type[E], bfs: bool = True) -> t.Optional[E]:
"""
Returns the first node in this tree which matches at least one of
the specified types.
Args:
expression_types: the expression type(s) to match.
bfs: whether to search the AST using the BFS algorithm (DFS is used if false).
Returns:
The node which matches the criteria or None if no such node was found.
"""
return next(self.find_all(*expression_types, bfs=bfs), None)
def find_all(self, *expression_types: t.Type[E], bfs: bool = True) -> t.Iterator[E]:
"""
Returns a generator object which visits all nodes in this tree and only
yields those that match at least one of the specified expression types.
Args:
expression_types: the expression type(s) to match.
bfs: whether to search the AST using the BFS algorithm (DFS is used if false).
Returns:
The generator object.
"""
for expression in self.walk(bfs=bfs):
if isinstance(expression, expression_types):
yield expression
def find_ancestor(self, *expression_types: t.Type[E]) -> t.Optional[E]:
"""
Returns a nearest parent matching expression_types.
Args:
expression_types: the expression type(s) to match.
Returns:
The parent node.
"""
ancestor = self.parent
while ancestor and not isinstance(ancestor, expression_types):
ancestor = ancestor.parent
return ancestor # type: ignore
@property
def parent_select(self) -> t.Optional[Select]:
"""
Returns the parent select statement.
"""
return self.find_ancestor(Select)
@property
def same_parent(self) -> bool:
"""Returns if the parent is the same class as itself."""
return type(self.parent) is self.__class__
def root(self) -> Expression:
"""
Returns the root expression of this tree.
"""
expression = self
while expression.parent:
expression = expression.parent
return expression
def walk(
self, bfs: bool = True, prune: t.Optional[t.Callable[[Expression], bool]] = None
) -> t.Iterator[Expression]:
"""
Returns a generator object which visits all nodes in this tree.
Args:
bfs: if set to True the BFS traversal order will be applied,
otherwise the DFS traversal will be used instead.
prune: callable that returns True if the generator should stop traversing
this branch of the tree.
Returns:
the generator object.
"""
if bfs:
yield from self.bfs(prune=prune)
else:
yield from self.dfs(prune=prune)
def dfs(
self, prune: t.Optional[t.Callable[[Expression], bool]] = None
) -> t.Iterator[Expression]:
"""
Returns a generator object which visits all nodes in this tree in
the DFS (Depth-first) order.
Returns:
The generator object.
"""
stack = [self]
while stack:
node = stack.pop()
yield node
if prune and prune(node):
continue
for v in node.iter_expressions(reverse=True):
stack.append(v)
def bfs(
self, prune: t.Optional[t.Callable[[Expression], bool]] = None
) -> t.Iterator[Expression]:
"""
Returns a generator object which visits all nodes in this tree in
the BFS (Breadth-first) order.
Returns:
The generator object.
"""
queue = deque([self])
while queue:
node = queue.popleft()
yield node
if prune and prune(node):
continue
for v in node.iter_expressions():
queue.append(v)
def unnest(self):
"""
Returns the first non parenthesis child or self.
"""
expression = self
while type(expression) is Paren:
expression = expression.this
return expression
def unalias(self):
"""
Returns the inner expression if this is an Alias.
"""
if isinstance(self, Alias):
return self.this
return self
def unnest_operands(self):
"""
Returns unnested operands as a tuple.
"""
return tuple(arg.unnest() for arg in self.iter_expressions())
def flatten(self, unnest=True):
"""
Returns a generator which yields child nodes whose parents are the same class.
A AND B AND C -> [A, B, C]
"""
for node in self.dfs(prune=lambda n: n.parent and type(n) is not self.__class__):
if type(node) is not self.__class__:
yield node.unnest() if unnest and not isinstance(node, Subquery) else node
def __str__(self) -> str:
return self.sql()
def __repr__(self) -> str:
return _to_s(self)
def to_s(self) -> str:
"""
Same as __repr__, but includes additional information which can be useful
for debugging, like empty or missing args and the AST nodes' object IDs.
"""
return _to_s(self, verbose=True)
def sql(self, dialect: DialectType = None, **opts) -> str:
"""
Returns SQL string representation of this tree.
Args:
dialect: the dialect of the output SQL string (eg. "spark", "hive", "presto", "mysql").
opts: other `sqlglot.generator.Generator` options.
Returns:
The SQL string.
"""
from sqlglot.dialects import Dialect
return Dialect.get_or_raise(dialect).generate(self, **opts)
def transform(self, fun: t.Callable, *args: t.Any, copy: bool = True, **kwargs) -> Expression:
"""
Visits all tree nodes (excluding already transformed ones)
and applies the given transformation function to each node.
Args:
fun: a function which takes a node as an argument and returns a
new transformed node or the same node without modifications. If the function
returns None, then the corresponding node will be removed from the syntax tree.
copy: if set to True a new tree instance is constructed, otherwise the tree is
modified in place.
Returns:
The transformed tree.
"""
root = None
new_node = None
for node in (self.copy() if copy else self).dfs(prune=lambda n: n is not new_node):
parent, arg_key, index = node.parent, node.arg_key, node.index
new_node = fun(node, *args, **kwargs)
if not root:
root = new_node
elif parent and arg_key and new_node is not node:
parent.set(arg_key, new_node, index)
assert root
return root.assert_is(Expression)
@t.overload
def replace(self, expression: E) -> E: ...
@t.overload
def replace(self, expression: None) -> None: ...
def replace(self, expression):
"""
Swap out this expression with a new expression.
For example::
>>> tree = Select().select("x").from_("tbl")
>>> tree.find(Column).replace(column("y"))
Column(
this=Identifier(this=y, quoted=False))
>>> tree.sql()
'SELECT y FROM tbl'
Args:
expression: new node
Returns:
The new expression or expressions.
"""
parent = self.parent
if not parent or parent is expression:
return expression
key = self.arg_key
value = parent.args.get(key)
if type(expression) is list and isinstance(value, Expression):
# We are trying to replace an Expression with a list, so it's assumed that
# the intention was to really replace the parent of this expression.
value.parent.replace(expression)
else:
parent.set(key, expression, self.index)
if expression is not self:
self.parent = None
self.arg_key = None
self.index = None
return expression
def pop(self: E) -> E:
"""
Remove this expression from its AST.
Returns:
The popped expression.
"""
self.replace(None)
return self
def assert_is(self, type_: t.Type[E]) -> E:
"""
Assert that this `Expression` is an instance of `type_`.
If it is NOT an instance of `type_`, this raises an assertion error.
Otherwise, this returns this expression.
Examples:
This is useful for type security in chained expressions:
>>> import sqlglot
>>> sqlglot.parse_one("SELECT x from y").assert_is(Select).select("z").sql()
'SELECT x, z FROM y'
"""
if not isinstance(self, type_):
raise AssertionError(f"{self} is not {type_}.")
return self
def error_messages(self, args: t.Optional[t.Sequence] = None) -> t.List[str]:
"""
Checks if this expression is valid (e.g. all mandatory args are set).
Args:
args: a sequence of values that were used to instantiate a Func expression. This is used
to check that the provided arguments don't exceed the function argument limit.
Returns:
A list of error messages for all possible errors that were found.
"""
errors: t.List[str] = []
if UNITTEST:
for k in self.args:
if k not in self.arg_types:
raise TypeError(f"Unexpected keyword: '{k}' for {self.__class__}")
for k in self.required_args:
v = self.args.get(k)
if v is None or (type(v) is list and not v):
errors.append(f"Required keyword: '{k}' missing for {self.__class__}")
if (
args
and isinstance(self, Func)
and len(args) > len(self.arg_types)
and not self.is_var_len_args
):
errors.append(
f"The number of provided arguments ({len(args)}) is greater than "
f"the maximum number of supported arguments ({len(self.arg_types)})"
)
return errors
def dump(self):
"""
Dump this Expression to a JSON-serializable dict.
"""
from sqlglot.serde import dump
return dump(self)
@classmethod
def load(cls, obj):
"""
Load a dict (as returned by `Expression.dump`) into an Expression instance.
"""
from sqlglot.serde import load
return load(obj)
def and_(
self,
*expressions: t.Optional[ExpOrStr],
dialect: DialectType = None,
copy: bool = True,
wrap: bool = True,
**opts,
) -> Condition:
"""
AND this condition with one or multiple expressions.
Example:
>>> condition("x=1").and_("y=1").sql()
'x = 1 AND y = 1'
Args:
*expressions: the SQL code strings to parse.
If an `Expression` instance is passed, it will be used as-is.
dialect: the dialect used to parse the input expression.
copy: whether to copy the involved expressions (only applies to Expressions).
wrap: whether to wrap the operands in `Paren`s. This is true by default to avoid
precedence issues, but can be turned off when the produced AST is too deep and
causes recursion-related issues.
opts: other options to use to parse the input expressions.
Returns:
The new And condition.
"""
return and_(self, *expressions, dialect=dialect, copy=copy, wrap=wrap, **opts)
def or_(
self,
*expressions: t.Optional[ExpOrStr],
dialect: DialectType = None,
copy: bool = True,
wrap: bool = True,
**opts,
) -> Condition:
"""
OR this condition with one or multiple expressions.
Example:
>>> condition("x=1").or_("y=1").sql()
'x = 1 OR y = 1'
Args:
*expressions: the SQL code strings to parse.
If an `Expression` instance is passed, it will be used as-is.
dialect: the dialect used to parse the input expression.
copy: whether to copy the involved expressions (only applies to Expressions).
wrap: whether to wrap the operands in `Paren`s. This is true by default to avoid
precedence issues, but can be turned off when the produced AST is too deep and
causes recursion-related issues.
opts: other options to use to parse the input expressions.
Returns:
The new Or condition.
"""
return or_(self, *expressions, dialect=dialect, copy=copy, wrap=wrap, **opts)
def not_(self, copy: bool = True):
"""
Wrap this condition with NOT.
Example:
>>> condition("x=1").not_().sql()
'NOT x = 1'
Args:
copy: whether to copy this object.
Returns:
The new Not instance.
"""
return not_(self, copy=copy)
def update_positions(
self: E,
other: t.Optional[Token | Expression] = None,
line: t.Optional[int] = None,
col: t.Optional[int] = None,
start: t.Optional[int] = None,
end: t.Optional[int] = None,
) -> E:
"""
Update this expression with positions from a token or other expression.
Args:
other: a token or expression to update this expression with.
line: the line number to use if other is None
col: column number
start: start char index
end: end char index
Returns:
The updated expression.
"""
if other is None:
self.meta["line"] = line
self.meta["col"] = col
self.meta["start"] = start
self.meta["end"] = end
elif hasattr(other, "meta"):
for k in POSITION_META_KEYS:
self.meta[k] = other.meta[k]
else:
self.meta["line"] = other.line
self.meta["col"] = other.col
self.meta["start"] = other.start
self.meta["end"] = other.end
return self
def as_(
self,
alias: str | Identifier,
quoted: t.Optional[bool] = None,
dialect: DialectType = None,
copy: bool = True,
**opts,
) -> Alias:
return alias_(self, alias, quoted=quoted, dialect=dialect, copy=copy, **opts)
def _binop(self, klass: t.Type[E], other: t.Any, reverse: bool = False) -> E:
this = self.copy()
other = convert(other, copy=True)
if not isinstance(this, klass) and not isinstance(other, klass):
this = _wrap(this, Binary)
other = _wrap(other, Binary)
if reverse:
return klass(this=other, expression=this)
return klass(this=this, expression=other)
def __getitem__(self, other: ExpOrStr | t.Tuple[ExpOrStr]) -> Bracket:
return Bracket(
this=self.copy(), expressions=[convert(e, copy=True) for e in ensure_list(other)]
)
def __iter__(self) -> t.Iterator:
if "expressions" in self.arg_types:
return iter(self.args.get("expressions") or [])
# We define this because __getitem__ converts Expression into an iterable, which is
# problematic because one can hit infinite loops if they do "for x in some_expr: ..."
# See: https://peps.python.org/pep-0234/
raise TypeError(f"'{self.__class__.__name__}' object is not iterable")
def isin(
self,
*expressions: t.Any,
query: t.Optional[ExpOrStr] = None,
unnest: t.Optional[ExpOrStr] | t.Collection[ExpOrStr] = None,
copy: bool = True,
**opts,
) -> In:
subquery = maybe_parse(query, copy=copy, **opts) if query else None
if subquery and not isinstance(subquery, Subquery):
subquery = subquery.subquery(copy=False)
return In(
this=maybe_copy(self, copy),
expressions=[convert(e, copy=copy) for e in expressions],
query=subquery,
unnest=(
Unnest(
expressions=[
maybe_parse(t.cast(ExpOrStr, e), copy=copy, **opts)
for e in ensure_list(unnest)
]
)
if unnest
else None
),
)
def between(
self,
low: t.Any,
high: t.Any,
copy: bool = True,
symmetric: t.Optional[bool] = None,
**opts,
) -> Between:
between = Between(
this=maybe_copy(self, copy),
low=convert(low, copy=copy, **opts),
high=convert(high, copy=copy, **opts),
)