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
|
import sys
from typing import Optional
from django.conf import settings
from django.core.management.base import BaseCommand
from django.utils.module_loading import import_string
from psqlextra.partitioning import PostgresPartitioningError
class Command(BaseCommand):
"""Create new partitions and delete old ones according to the configured
partitioning strategies."""
help = "Create new partitions and delete old ones using the configured partitioning manager. The PSQLEXTRA_PARTITIONING_MANAGER setting must be configured."
def add_arguments(self, parser):
parser.add_argument(
"--dry",
"-d",
action="store_true",
help="When specified, no partition will be created/deleted. Just a simulation.",
required=False,
default=False,
)
parser.add_argument(
"--yes",
"-y",
action="store_true",
help="Answer yes to all questions. WARNING: You will not be asked before deleting a partition.",
required=False,
default=False,
)
parser.add_argument(
"--using",
"-u",
help="Optional name of the database connection to use.",
default="default",
)
parser.add_argument(
"--skip-create",
action="store_true",
help="Do not create partitions.",
required=False,
default=False,
)
parser.add_argument(
"--skip-delete",
action="store_true",
help="Do not delete partitions.",
required=False,
default=False,
)
def handle( # type: ignore[override]
self,
dry: bool,
yes: bool,
using: Optional[str],
skip_create: bool,
skip_delete: bool,
*args,
**kwargs,
):
partitioning_manager = self._partitioning_manager()
plan = partitioning_manager.plan(
skip_create=skip_create, skip_delete=skip_delete, using=using
)
creations_count = len(plan.creations)
deletions_count = len(plan.deletions)
if creations_count == 0 and deletions_count == 0:
print("Nothing to be done.")
return
plan.print()
if dry:
return
if not yes:
sys.stdout.write("Do you want to proceed? (y/N) ")
if not self._ask_for_confirmation():
print("Operation aborted.")
return
plan.apply(using=using)
print("Operations applied.")
@staticmethod
def _ask_for_confirmation() -> bool:
answer = input("").lower()
if not answer:
return False
if answer[0] == "y" or answer == "yes":
return True
return False
@staticmethod
def _partitioning_manager():
partitioning_manager = getattr(
settings, "PSQLEXTRA_PARTITIONING_MANAGER", None
)
if not partitioning_manager:
raise PostgresPartitioningError(
"You must configure the PSQLEXTRA_PARTITIONING_MANAGER setting "
"for automatic partitioning to work."
)
if isinstance(partitioning_manager, str):
partitioning_manager = import_string(partitioning_manager)
return partitioning_manager
|
