Coverage for /var/srv/projects/api.amasfac.comuna18.com/tmp/venv/lib/python3.9/site-packages/django/core/management/commands/createcachetable.py: 27%

1from django.conf import settings 

2from django.core.cache import caches 

3from django.core.cache.backends.db import BaseDatabaseCache 

4from django.core.management.base import BaseCommand, CommandError 

5from django.db import ( 

6 DEFAULT_DB_ALIAS, 

7 DatabaseError, 

8 connections, 

9 models, 

10 router, 

11 transaction, 

12) 

13 

14 

15class Command(BaseCommand): 

16 help = "Creates the tables needed to use the SQL cache backend." 

17 

18 requires_system_checks = [] 

19 

20 def add_arguments(self, parser): 

21 parser.add_argument( 

22 "args", 

23 metavar="table_name", 

24 nargs="*", 

25 help=( 

26 "Optional table names. Otherwise, settings.CACHES is used to find " 

27 "cache tables." 

28 ), 

29 ) 

30 parser.add_argument( 

31 "--database", 

32 default=DEFAULT_DB_ALIAS, 

33 help="Nominates a database onto which the cache tables will be " 

34 'installed. Defaults to the "default" database.', 

35 ) 

36 parser.add_argument( 

37 "--dry-run", 

38 action="store_true", 

39 help="Does not create the table, just prints the SQL that would be run.", 

40 ) 

41 

42 def handle(self, *tablenames, **options): 

43 db = options["database"] 

44 self.verbosity = options["verbosity"] 

45 dry_run = options["dry_run"] 

46 if tablenames: 46 ↛ 48line 46 didn't jump to line 48, because the condition on line 46 was never true

47 # Legacy behavior, tablename specified as argument 

48 for tablename in tablenames: 

49 self.create_table(db, tablename, dry_run) 

50 else: 

51 for cache_alias in settings.CACHES: 

52 cache = caches[cache_alias] 

53 if isinstance(cache, BaseDatabaseCache): 53 ↛ 54line 53 didn't jump to line 54, because the condition on line 53 was never true

54 self.create_table(db, cache._table, dry_run) 

55 

56 def create_table(self, database, tablename, dry_run): 

57 cache = BaseDatabaseCache(tablename, {}) 

58 if not router.allow_migrate_model(database, cache.cache_model_class): 

59 return 

60 connection = connections[database] 

61 

62 if tablename in connection.introspection.table_names(): 

63 if self.verbosity > 0: 

64 self.stdout.write("Cache table '%s' already exists." % tablename) 

65 return 

66 

67 fields = ( 

68 # "key" is a reserved word in MySQL, so use "cache_key" instead. 

69 models.CharField( 

70 name="cache_key", max_length=255, unique=True, primary_key=True 

71 ), 

72 models.TextField(name="value"), 

73 models.DateTimeField(name="expires", db_index=True), 

74 ) 

75 table_output = [] 

76 index_output = [] 

77 qn = connection.ops.quote_name 

78 for f in fields: 

79 field_output = [ 

80 qn(f.name), 

81 f.db_type(connection=connection), 

82 "%sNULL" % ("NOT " if not f.null else ""), 

83 ] 

84 if f.primary_key: 

85 field_output.append("PRIMARY KEY") 

86 elif f.unique: 

87 field_output.append("UNIQUE") 

88 if f.db_index: 

89 unique = "UNIQUE " if f.unique else "" 

90 index_output.append( 

91 "CREATE %sINDEX %s ON %s (%s);" 

92 % ( 

93 unique, 

94 qn("%s_%s" % (tablename, f.name)), 

95 qn(tablename), 

96 qn(f.name), 

97 ) 

98 ) 

99 table_output.append(" ".join(field_output)) 

100 full_statement = ["CREATE TABLE %s (" % qn(tablename)] 

101 for i, line in enumerate(table_output): 

102 full_statement.append( 

103 " %s%s" % (line, "," if i < len(table_output) - 1 else "") 

104 ) 

105 full_statement.append(");") 

106 

107 full_statement = "\n".join(full_statement) 

108 

109 if dry_run: 

110 self.stdout.write(full_statement) 

111 for statement in index_output: 

112 self.stdout.write(statement) 

113 return 

114 

115 with transaction.atomic( 

116 using=database, savepoint=connection.features.can_rollback_ddl 

117 ): 

118 with connection.cursor() as curs: 

119 try: 

120 curs.execute(full_statement) 

121 except DatabaseError as e: 

122 raise CommandError( 

123 "Cache table '%s' could not be created.\nThe error was: %s." 

124 % (tablename, e) 

125 ) 

126 for statement in index_output: 

127 curs.execute(statement) 

128 

129 if self.verbosity > 1: 

130 self.stdout.write("Cache table '%s' created." % tablename)