-
Notifications
You must be signed in to change notification settings - Fork 5
/
Copy pathdb.py
46 lines (34 loc) · 1.26 KB
/
db.py
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
"""
NOTE: on using peewee for non-integer primary_key
```python
# This works because .create() will specify `force_insert=True`.
obj1 = UUIDModel.create(id=uuid.uuid4())
# This will not work, however. Peewee will attempt to do an update:
obj2 = UUIDModel(id=uuid.uuid4())
obj2.save() # WRONG
obj2.save(force_insert=True) # CORRECT
# Once the object has been created, you can call save() normally.
obj2.save()
```
Read more: http://docs.peewee-orm.com/en/latest/peewee/models.html?highlight=force_insert#id4
"""
from baca.exceptions import TableDoesNotExist
from baca.models import DbMetadata, Migration, ReadingHistory, db
def initial_migration() -> None:
db.create_tables([DbMetadata, ReadingHistory])
MIGRATIONS: list[Migration] = [
Migration(version=0, migrate=initial_migration),
]
def migrate() -> None:
db.connect()
try:
for migration in sorted(MIGRATIONS, key=lambda x: x.version):
try:
if not DbMetadata.table_exists():
raise TableDoesNotExist()
DbMetadata.get_by_id(migration.version)
except (DbMetadata.DoesNotExist, TableDoesNotExist):
migration.migrate()
DbMetadata.create(version=migration.version)
finally:
db.close()