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
|
"""Added oauth models
Revision ID: fc6608d3f9cc
Revises: 115caa5a76d0
Create Date: 2026-07-02 17:40:58.968553
"""
import json
import os
from collections.abc import Sequence
from datetime import UTC, datetime
from secrets import token_hex
from uuid import uuid4, uuid7
import sqlalchemy as sa
import sqlmodel
from alembic import op
# revision identifiers, used by Alembic.
revision: str = "fc6608d3f9cc"
down_revision: str | Sequence[str] | None = "115caa5a76d0"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
client_application = op.create_table(
"clientapplication",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("name", sqlmodel.sql.sqltypes.AutoString(), nullable=False),
sa.Column("client_id", sa.String(length=48), nullable=True),
sa.Column("client_secret", sa.String(length=120), nullable=True),
sa.Column("client_id_issued_at", sa.Integer(), nullable=False),
sa.Column("client_secret_expires_at", sa.Integer(), nullable=False),
sa.Column("client_metadata", sa.Text(), nullable=True),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(
op.f("ix_clientapplication_client_id"),
"clientapplication",
["client_id"],
unique=False,
)
op.create_table(
"authcode",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("user_id", sa.Uuid(), nullable=False),
sa.Column("code", sa.String(length=120), nullable=False),
sa.Column("client_id", sa.String(length=48), nullable=True),
sa.Column("redirect_uri", sa.Text(), nullable=True),
sa.Column("response_type", sa.Text(), nullable=True),
sa.Column("scope", sa.Text(), nullable=True),
sa.Column("nonce", sa.Text(), nullable=True),
sa.Column("auth_time", sa.Integer(), nullable=False),
sa.Column("acr", sa.Text(), nullable=True),
sa.Column("amr", sa.Text(), nullable=True),
sa.Column("code_challenge", sa.Text(), nullable=True),
sa.Column("code_challenge_method", sa.String(length=48), nullable=True),
sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("code"),
)
op.create_table(
"authtoken",
sa.Column("id", sa.Uuid(), nullable=False),
sa.Column("user_id", sa.Uuid(), nullable=False),
sa.Column("client_id", sa.String(length=48), nullable=True),
sa.Column("token_type", sa.String(length=40), nullable=True),
sa.Column("access_token", sa.String(length=255), nullable=False),
sa.Column("refresh_token", sa.String(length=255), nullable=True),
sa.Column("scope", sa.Text(), nullable=True),
sa.Column("issued_at", sa.Integer(), nullable=False),
sa.Column("access_token_revoked_at", sa.Integer(), nullable=False),
sa.Column("refresh_token_revoked_at", sa.Integer(), nullable=False),
sa.Column("expires_in", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(["user_id"], ["user.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("access_token"),
)
op.create_index(
op.f("ix_authtoken_refresh_token"), "authtoken", ["refresh_token"], unique=False
)
op.bulk_insert(
client_application,
[
{
"id": uuid7(),
"name": "TTUN",
"client_id": uuid4().hex,
"client_secret": token_hex(32),
"client_id_issued_at": datetime.now(tz=UTC).timestamp(),
"client_secret_expires_at": datetime.now(tz=UTC).timestamp() + 3600,
"client_metadata": json.dumps(
{
"redirect_uris": [
f"//{os.environ['TUNNEL_DOMAIN']}/oauth/callback"
],
}
),
}
],
)
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f("ix_authtoken_refresh_token"), table_name="authtoken")
op.drop_table("authtoken")
op.drop_table("authcode")
op.drop_index(
op.f("ix_clientapplication_client_id"), table_name="clientapplication"
)
op.drop_table("clientapplication")
# ### end Alembic commands ###
|
