Coverage for packages/sql-smith/src/sql_smith/engine/basic_engine.py: 88%

1from functools import reduce 

2from typing import Callable 

3 

4from sql_smith.interfaces import EngineInterface, StatementInterface 

5from sql_smith.query import DeleteQuery, InsertQuery, SelectQuery, UpdateQuery 

6 

7 

8class BasicEngine(EngineInterface): 

9 """Basic engine 

10 

11 A default implementation of an engine. 

12 """ 

13 

14 def make_select(self) -> "SelectQuery": 

15 """Creates a SelectQuery.""" 

16 return SelectQuery(self) 

17 

18 def make_insert(self) -> "InsertQuery": 

19 """Creates an InsertQuery.""" 

20 return InsertQuery(self) 

21 

22 def make_update(self) -> "UpdateQuery": 

23 """Creates an UpdateQuery.""" 

24 return UpdateQuery(self) 

25 

26 def make_delete(self) -> "DeleteQuery": 

27 """Creates a DeleteQuery.""" 

28 return DeleteQuery(self) 

29 

30 def escape_identifier(self, identifier: str) -> str: 

31 """Escapes an identifier. 

32 

33 The default implementation returns the identifier as is. 

34 """ 

35 return identifier 

36 

37 def escape_like(self, parameter: str) -> str: 

38 """Escapes the LIKE argument 

39 

40 A backslash is used to escape wildcards. 

41 Standard wildcards are underscore and percent sign. 

42 """ 

43 return parameter.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") 

44 

45 def export_parameter(self, param) -> str: 

46 """Export a parameter. 

47 

48 A boolean will be exported as true or false. A None value will be exported as NULL. 

49 """ 

50 if isinstance(param, bool): 

51 return "true" if param else "false" 

52 if param is None: 

53 return "NULL" 

54 return str(param) 

55 

56 def extract_params(self) -> Callable[["StatementInterface"], tuple]: 

57 """Create a lambda to extract parameters.""" 

58 return lambda statement: statement.params(self) 

59 

60 def extract_sql(self) -> Callable[["StatementInterface"], str]: 

61 """Create a lambda to extract SQL.""" 

62 return lambda statement: statement.sql(self) 

63 

64 def flatten_params(self, *args: "StatementInterface") -> tuple: 

65 """Create a tuple with all parameters found.""" 

66 # map with extract_params returns a list with tuples, 

67 # the reduce method is used to merge all these tuples into one big tuple 

68 return reduce(lambda x, y: (x + y), map(self.extract_params(), args), ()) 

69 

70 def flatten_sql(self, separator: str, *args: "StatementInterface") -> str: 

71 """Transform StatementInterface arguments to a string.""" 

72 return separator.join(list(map(self.extract_sql(), args))) 

73 

74 def get_parameter_placeholder(self) -> str: 

75 return "?"