blob: a60a12197f5cea3ba80f68fe5359672c3ce640bf (
plain) (
blame)
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
|
import sqlite3
from pathlib import Path
from shutil import copyfile
from bot.seasons.season import get_seasons
DIRECTORY = Path("data") # directory that has a persistent volume mapped to it
def make_persistent(file_path: Path) -> Path:
"""
Copy datafile at the provided file_path to the persistent data directory.
A persistent data file is needed by some features in order to not lose data
after bot rebuilds.
This function will ensure that a clean data file with default schema,
structure or data is copied over to the persistent volume before returning
the path to this new persistent version of the file.
If the persistent file already exists, it won't be overwritten with the
clean default file, just returning the Path instead to the existing file.
Note: Avoid using the same file name as other features in the same seasons
as otherwise only one datafile can be persistent and will be returned for
both cases.
Example Usage:
>>> import json
>>> template_datafile = Path("bot", "resources", "evergreen", "myfile.json")
>>> path_to_persistent_file = make_persistent(template_datafile)
>>> print(path_to_persistent_file)
data/evergreen/myfile.json
>>> with path_to_persistent_file.open("w+") as f:
>>> data = json.load(f)
"""
# ensure the persistent data directory exists
DIRECTORY.mkdir(exist_ok=True)
if not file_path.is_file():
raise OSError(f"File not found at {file_path}.")
# detect season in datafile path for assigning to subdirectory
season = next((s for s in get_seasons() if s in file_path.parts), None)
if season:
# make sure subdirectory exists first
subdirectory = Path(DIRECTORY, season)
subdirectory.mkdir(exist_ok=True)
persistent_path = Path(subdirectory, file_path.name)
else:
persistent_path = Path(DIRECTORY, file_path.name)
# copy base/template datafile to persistent directory
if not persistent_path.exists():
copyfile(file_path, persistent_path)
return persistent_path
def sqlite(db_path: Path) -> sqlite3.Connection:
"""Copy sqlite file to the persistent data directory and return an open connection."""
persistent_path = make_persistent(db_path)
return sqlite3.connect(persistent_path)
|
