-
Notifications
You must be signed in to change notification settings - Fork 1
/
Copy pathmanage.py
155 lines (112 loc) · 3.81 KB
/
manage.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
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
from typing import Annotated, Any
import typer
import uvicorn
from alembic.config import Config
from app.core.config import config
app = typer.Typer()
@app.command(help="Show current revision.")
def currentrevision():
"""
Show current revision.
This function shows the current revision of the database using Alembic.
"""
from alembic.command import current
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
alembic_cfg.set_main_option("sqlalchemy.url", db_url())
current(alembic_cfg)
@app.command(help="Prints the migration history using Alembic.")
def migrationshistory():
"""
Prints the migration history using Alembic.
Returns:
None
"""
from alembic.command import history
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
history(alembic_cfg)
@app.command(help="Create a new Alembic migration.")
def makemigrations(m: Annotated[str, typer.Option(help="The migration message.")] = ""):
"""
Create a new Alembic migration.
Args:
m: The migration message.
Returns:
None
"""
from alembic.command import revision
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
alembic_cfg.set_main_option("sqlalchemy.url", db_url())
revision_kwargs: dict[str, Any] = {"autogenerate": True}
revision_kwargs["message"] = m
revision(alembic_cfg, **revision_kwargs)
@app.command(help="Upgrade the database schema to the latest version.")
def migrate(
revision: Annotated[
str, typer.Option(help="The revision to upgrade to. Defaults to 'head'.")
] = "head"
):
"""
Upgrade the database schema to the latest version.
Args:
revision: The revision to upgrade to. Defaults to "head".
"""
from alembic.command import upgrade
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
alembic_cfg.set_main_option("sqlalchemy.url", db_url())
upgrade(alembic_cfg, revision)
@app.command(help="Downgrade the database schema by a specified number of revisions.")
def downgrade(
step: Annotated[
int, typer.Option(help="The number of revisions to downgrade. Defaults to 1.")
] = 1
):
"""
Downgrade the database schema by a specified number of revisions.
Args:
step: The number of revisions to downgrade. Defaults to 1.
Returns:
None
"""
from alembic.command import downgrade
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
alembic_cfg.set_main_option("sqlalchemy.url", db_url())
downgrade(alembic_cfg, f"-{step}")
@app.command(help="Stamp the revision of the database schema.")
def stamp(revision: Annotated[str, typer.Option(help="The revision number to stamp.")]):
"""
Stamp the revision of the database schema.
Args:
revision: The revision number to stamp.
Returns:
None
"""
from alembic.command import stamp
alembic_ini_path = "./alembic.ini"
alembic_cfg = Config(alembic_ini_path)
alembic_cfg.set_main_option("sqlalchemy.url", db_url())
stamp(alembic_cfg, revision)
@app.command(help="Runs the server.")
def runserver(
host: Annotated[
str, typer.Option(help="The host to bind the server to. Defaults to '0.0.0.0'.")
] = "0.0.0.0",
port: Annotated[
int, typer.Option(help="The port to bind the server to. Defaults to 8000.")
] = 8000,
):
"""
Runs the server
Args:
host: The host to bind the server to. Defaults to "0.0.0.0".
port: The port to bind the server to. Defaults to 8000.
"""
uvicorn.run("app.main:app", host=host, port=port, reload=True) # pyright: ignore
def db_url():
return str(config.db_url).replace("+asyncpg", "")
if __name__ == "__main__":
app()