Bennett/g0v0-server
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' of https://github.com/GooGuTeam/g0v0-server

Browse Source
This commit is contained in:
咕谷酱
2025-08-22 14:03:00 +08:00
parent 3350081e80 08e899c140
commit c8a68f97ed
11 changed files with 532 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
app/achievements/hush_hush.py
View File

@@ -379,7 +379,7 @@ async def deliberation(
return False
if (
not beatmap.beatmap_status.has_pp()
or beatmap.beatmap_status != BeatmapRankStatus.LOVED
and beatmap.beatmap_status != BeatmapRankStatus.LOVED
):
return False

2
app/achievements/mods.py
View File

@@ -34,6 +34,8 @@ async def process_category_mod(
return False
if not beatmap.beatmap_status.has_leaderboard():
return False
if len(score.mods) == 0:
return False
api_mods = {
k
for k, v in API_MODS[int(score.gamemode)].items() # pyright: ignore[reportArgumentType]

7
app/database/__init__.py
View File

@@ -31,6 +31,7 @@ from .lazer_user import (
)
from .multiplayer_event import MultiplayerEvent, MultiplayerEventResp
from .notification import Notification, UserNotification
from .password_reset import PasswordReset
from .playlist_attempts import (
ItemAttemptsCount,
ItemAttemptsResp,
@@ -56,7 +57,7 @@ from .statistics import (
UserStatistics,
UserStatisticsResp,
)
from .team import Team, TeamMember
from .team import Team, TeamMember, TeamRequest
from .user_account_history import (
UserAccountHistory,
UserAccountHistoryResp,
@@ -81,9 +82,11 @@ __all__ = [
"CountResp",
"DailyChallengeStats",
"DailyChallengeStatsResp",
"EmailVerification",
"FavouriteBeatmapset",
"ItemAttemptsCount",
"ItemAttemptsResp",
"LoginSession",
"MonthlyPlaycounts",
"MultiplayerEvent",
"MultiplayerEventResp",
@@ -92,6 +95,7 @@ __all__ = [
"OAuthClient",
"OAuthToken",
"PPBestScore",
"PasswordReset",
"Playlist",
"PlaylistAggregateScore",
"PlaylistBestScore",
@@ -115,6 +119,7 @@ __all__ = [
"ScoreTokenResp",
"Team",
"TeamMember",
"TeamRequest",
"User",
"UserAccountHistory",
"UserAccountHistoryResp",

31
app/database/team.py
View File

@@ -1,4 +1,4 @@
from datetime import datetime
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from app.models.model import UTCBaseModel
@@ -16,19 +16,21 @@ class Team(SQLModel, UTCBaseModel, table=True):
id: int | None = Field(default=None, primary_key=True, index=True)
name: str = Field(max_length=100)
short_name: str = Field(max_length=10)
flag_url: str | None = Field(default=None, max_length=500)
created_at: datetime = Field(
default_factory=datetime.utcnow, sa_column=Column(DateTime)
)
flag_url: str | None = Field(default=None)
cover_url: str | None = Field(default=None)
created_at: datetime = Field(default=datetime.now(UTC), sa_column=Column(DateTime))
leader_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id")))
leader: "User" = Relationship()
members: list["TeamMember"] = Relationship(back_populates="team")
class TeamMember(SQLModel, UTCBaseModel, table=True):
__tablename__ = "team_members" # pyright: ignore[reportAssignmentType]
id: int | None = Field(default=None, primary_key=True, index=True)
user_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id")))
user_id: int = Field(
sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), primary_key=True)
)
team_id: int = Field(foreign_key="teams.id")
joined_at: datetime = Field(
default_factory=datetime.utcnow, sa_column=Column(DateTime)
@@ -40,3 +42,18 @@ class TeamMember(SQLModel, UTCBaseModel, table=True):
team: "Team" = Relationship(
back_populates="members", sa_relationship_kwargs={"lazy": "joined"}
)
class TeamRequest(SQLModel, UTCBaseModel, table=True):
__tablename__ = "team_requests" # pyright: ignore[reportAssignmentType]
user_id: int = Field(
sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), primary_key=True)
)
team_id: int = Field(foreign_key="teams.id", primary_key=True)
requested_at: datetime = Field(
default=datetime.now(UTC), sa_column=Column(DateTime)
)
user: "User" = Relationship(sa_relationship_kwargs={"lazy": "joined"})
team: "Team" = Relationship(sa_relationship_kwargs={"lazy": "joined"})

91
app/models/notification.py
View File

@@ -2,7 +2,7 @@ from __future__ import annotations
from abc import abstractmethod
from enum import Enum
from typing import TYPE_CHECKING, Self
from typing import TYPE_CHECKING, ClassVar, Self
from app.utils import truncate
@@ -16,7 +16,7 @@ from sqlmodel.ext.asyncio.session import AsyncSession
CONTENT_TRUNCATE = 36
if TYPE_CHECKING:
from app.database import ChannelType, ChatMessage, User
from app.database import ChannelType, ChatMessage, TeamRequest, User
# https://github.com/ppy/osu-web/blob/master/app/Models/Notification.php
@@ -78,10 +78,7 @@ class NotificationName(str, Enum):
class NotificationDetail(BaseModel):
@property
@abstractmethod
def name(self) -> NotificationName:
raise NotImplementedError
name: ClassVar[NotificationName]
@property
@abstractmethod
@@ -104,9 +101,9 @@ class NotificationDetail(BaseModel):
class ChannelMessageBase(NotificationDetail):
title: str = ""
type: str = ""
cover_url: str = ""
title: str
type: str
cover_url: str
_message: "ChatMessage" = PrivateAttr()
_user: "User" = PrivateAttr()
@@ -147,9 +144,7 @@ class ChannelMessageBase(NotificationDetail):
class ChannelMessage(ChannelMessageBase):
@property
def name(self) -> NotificationName:
return NotificationName.CHANNEL_MESSAGE
name: ClassVar[NotificationName] = NotificationName.CHANNEL_MESSAGE
class ChannelMessageTeam(ChannelMessageBase):
@@ -163,9 +158,7 @@ class ChannelMessageTeam(ChannelMessageBase):
return super().init(message, user, [], ChannelType.TEAM)
@property
def name(self) -> NotificationName:
return NotificationName.CHANNEL_TEAM
name: ClassVar[NotificationName] = NotificationName.CHANNEL_TEAM
async def get_receivers(self, session: AsyncSession) -> list[int]:
from app.database import TeamMember
@@ -210,9 +203,7 @@ class UserAchievementUnlock(NotificationDetail):
async def get_receivers(self, session: AsyncSession) -> list[int]:
return [self.user_id]
@property
def name(self) -> NotificationName:
return NotificationName.USER_ACHIEVEMENT_UNLOCK
name: ClassVar[NotificationName] = NotificationName.USER_ACHIEVEMENT_UNLOCK
@property
def object_id(self) -> int:
@@ -227,4 +218,66 @@ class UserAchievementUnlock(NotificationDetail):
return "achievement"
NotificationDetails = ChannelMessage | ChannelMessageTeam | UserAchievementUnlock
class TeamApplicationBase(NotificationDetail):
cover_url: str
title: str
_team_request: "TeamRequest" = PrivateAttr()
@classmethod
def init(cls, team_request: "TeamRequest") -> Self:
instance = cls(
title=team_request.team.name,
cover_url=team_request.team.flag_url or "",
)
instance._team_request = team_request
return instance
async def get_receivers(self, session: AsyncSession) -> list[int]:
return [self._team_request.user_id]
@property
def object_id(self) -> int:
return self._team_request.team_id
@property
def source_user_id(self) -> int:
return self._team_request.user_id
@property
def object_type(self) -> str:
return "team"
class TeamApplicationAccept(TeamApplicationBase):
name: ClassVar[NotificationName] = NotificationName.TEAM_APPLICATION_ACCEPT
class TeamApplicationReject(TeamApplicationBase):
name: ClassVar[NotificationName] = NotificationName.TEAM_APPLICATION_REJECT
class TeamApplicationStore(TeamApplicationBase):
name: ClassVar[NotificationName] = NotificationName.TEAM_APPLICATION_STORE
async def get_receivers(self, session: AsyncSession) -> list[int]:
return [self._team_request.team.leader_id]
@classmethod
def init(cls, team_request: "TeamRequest") -> Self:
instance = cls(
title=team_request.user.username,
cover_url=team_request.team.flag_url or "",
)
instance._team_request = team_request
return instance
NotificationDetails = (
ChannelMessage
| ChannelMessageTeam
| UserAchievementUnlock
| TeamApplicationAccept
| TeamApplicationReject
| TeamApplicationStore
)

2
app/router/private/__init__.py
View File

@@ -1,6 +1,6 @@
from __future__ import annotations
from . import avatar, cover, oauth, relationship, username # noqa: F401
from . import avatar, cover, oauth, relationship, team, username # noqa: F401
from .router import router as private_router
__all__ = [

20
app/router/private/avatar.py
View File

@@ -1,18 +1,17 @@
from __future__ import annotations
import hashlib
from io import BytesIO
from app.database.lazer_user import User
from app.dependencies.database import Database
from app.dependencies.storage import get_storage_service
from app.dependencies.user import get_client_user
from app.storage.base import StorageService
from app.utils import check_image
from .router import router
from fastapi import Depends, File, HTTPException, Security
from PIL import Image
from fastapi import Depends, File, Security
@router.post(
@@ -39,20 +38,7 @@ async def upload_avatar(
"""
# check file
if len(content) > 5 * 1024 * 1024: # 5MB limit
raise HTTPException(status_code=400, detail="File size exceeds 5MB limit")
elif len(content) == 0:
raise HTTPException(status_code=400, detail="File cannot be empty")
try:
with Image.open(BytesIO(content)) as img:
if img.format not in ["PNG", "JPEG", "GIF"]:
raise HTTPException(status_code=400, detail="Invalid image format")
if img.size[0] > 256 or img.size[1] > 256:
raise HTTPException(
status_code=400, detail="Image size exceeds 256x256 pixels"
)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error processing image: {e}")
check_image(content, 5 * 1024 * 1024, 256, 256)
filehash = hashlib.sha256(content).hexdigest()
storage_path = f"avatars/{current_user.id}_{filehash}.png"

20
app/router/private/cover.py
View File

@@ -1,18 +1,17 @@
from __future__ import annotations
import hashlib
from io import BytesIO
from app.database.lazer_user import User, UserProfileCover
from app.dependencies.database import Database
from app.dependencies.storage import get_storage_service
from app.dependencies.user import get_client_user
from app.storage.base import StorageService
from app.utils import check_image
from .router import router
from fastapi import Depends, File, HTTPException, Security
from PIL import Image
from fastapi import Depends, File, Security
@router.post(
@@ -39,20 +38,7 @@ async def upload_cover(
"""
# check file
if len(content) > 10 * 1024 * 1024: # 10MB limit
raise HTTPException(status_code=400, detail="File size exceeds 10MB limit")
elif len(content) == 0:
raise HTTPException(status_code=400, detail="File cannot be empty")
try:
with Image.open(BytesIO(content)) as img:
if img.format not in ["PNG", "JPEG", "GIF"]:
raise HTTPException(status_code=400, detail="Invalid image format")
if img.size[0] > 3000 or img.size[1] > 2000:
raise HTTPException(
status_code=400, detail="Image size exceeds 3000x2000 pixels"
)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error processing image: {e}")
check_image(content, 10 * 1024 * 1024, 3000, 2000)
filehash = hashlib.sha256(content).hexdigest()
storage_path = f"cover/{current_user.id}_{filehash}.png"

312
app/router/private/team.py Normal file
View File

@@ -0,0 +1,312 @@
from __future__ import annotations
from datetime import UTC, datetime
import hashlib
from app.database.lazer_user import BASE_INCLUDES, User, UserResp
from app.database.team import Team, TeamMember, TeamRequest
from app.dependencies.database import Database
from app.dependencies.storage import get_storage_service
from app.dependencies.user import get_client_user
from app.models.notification import (
TeamApplicationAccept,
TeamApplicationReject,
TeamApplicationStore,
)
from app.router.notification import server
from app.storage.base import StorageService
from app.utils import check_image
from .router import router
from fastapi import Depends, File, Form, HTTPException, Path, Request, Security
from pydantic import BaseModel
from sqlmodel import exists, select
@router.post("/team", name="创建战队", response_model=Team)
async def create_team(
session: Database,
storage: StorageService = Depends(get_storage_service),
current_user: User = Security(get_client_user),
flag: bytes = File(..., description="战队图标文件"),
cover: bytes = File(..., description="战队头图文件"),
name: str = Form(max_length=100, description="战队名称"),
short_name: str = Form(max_length=10, description="战队缩写"),
):
"""创建战队。
flag 限制 240x120, 2MB; cover 限制 3000x2000, 10MB
支持的图片格式: PNG、JPEG、GIF
"""
user_id = current_user.id
assert user_id
if (await current_user.awaitable_attrs.team_membership) is not None:
raise HTTPException(status_code=403, detail="You are already in a team")
is_existed = (await session.exec(select(exists()).where(Team.name == name))).first()
if is_existed:
raise HTTPException(status_code=409, detail="Name already exists")
is_existed = (
await session.exec(select(exists()).where(Team.short_name == short_name))
).first()
if is_existed:
raise HTTPException(status_code=409, detail="Short name already exists")
check_image(flag, 2 * 1024 * 1024, 240, 120)
check_image(cover, 10 * 1024 * 1024, 3000, 2000)
now = datetime.now(UTC)
team = Team(name=name, short_name=short_name, leader_id=user_id, created_at=now)
session.add(team)
await session.commit()
await session.refresh(team)
filehash = hashlib.sha256(flag).hexdigest()
storage_path = f"team_flag/{team.id}_{filehash}.png"
if not await storage.is_exists(storage_path):
await storage.write_file(storage_path, flag)
team.flag_url = await storage.get_file_url(storage_path)
filehash = hashlib.sha256(cover).hexdigest()
storage_path = f"team_cover/{team.id}_{filehash}.png"
if not await storage.is_exists(storage_path):
await storage.write_file(storage_path, cover)
team.cover_url = await storage.get_file_url(storage_path)
team_member = TeamMember(user_id=user_id, team_id=team.id, joined_at=now)
session.add(team_member)
await session.commit()
await session.refresh(team)
return team
@router.patch("/team/{team_id}", name="修改战队", response_model=Team)
async def update_team(
team_id: int,
session: Database,
storage: StorageService = Depends(get_storage_service),
current_user: User = Security(get_client_user),
flag: bytes | None = File(default=None, description="战队图标文件"),
cover: bytes | None = File(default=None, description="战队头图文件"),
name: str | None = Form(default=None, max_length=100, description="战队名称"),
short_name: str | None = Form(default=None, max_length=10, description="战队缩写"),
leader_id: int | None = Form(default=None, description="战队队长 ID"),
):
"""修改战队。
flag 限制 240x120, 2MB; cover 限制 3000x2000, 10MB
支持的图片格式: PNG、JPEG、GIF
"""
team = await session.get(Team, team_id)
user_id = current_user.id
assert user_id
if not team:
raise HTTPException(status_code=404, detail="Team not found")
if team.leader_id != user_id:
raise HTTPException(status_code=403, detail="You are not the team leader")
is_existed = (await session.exec(select(exists()).where(Team.name == name))).first()
if is_existed:
raise HTTPException(status_code=409, detail="Name already exists")
is_existed = (
await session.exec(select(exists()).where(Team.short_name == short_name))
).first()
if is_existed:
raise HTTPException(status_code=409, detail="Short name already exists")
if flag:
check_image(flag, 2 * 1024 * 1024, 240, 120)
filehash = hashlib.sha256(flag).hexdigest()
storage_path = f"team_flag/{team.id}_{filehash}.png"
if not await storage.is_exists(storage_path):
await storage.write_file(storage_path, flag)
team.flag_url = await storage.get_file_url(storage_path)
if cover:
check_image(cover, 10 * 1024 * 1024, 3000, 2000)
filehash = hashlib.sha256(cover).hexdigest()
storage_path = f"team_cover/{team.id}_{filehash}.png"
if not await storage.is_exists(storage_path):
await storage.write_file(storage_path, cover)
team.cover_url = await storage.get_file_url(storage_path)
if leader_id is not None:
if not (
await session.exec(select(exists()).where(User.id == leader_id))
).first():
raise HTTPException(status_code=404, detail="Leader not found")
if not (
await session.exec(
select(TeamMember).where(
TeamMember.user_id == leader_id, TeamMember.team_id == team.id
)
)
).first():
raise HTTPException(
status_code=404, detail="Leader is not a member of the team"
)
team.leader_id = leader_id
await session.commit()
await session.refresh(team)
return team
@router.delete("/team/{team_id}", name="删除战队", status_code=204)
async def delete_team(
session: Database,
team_id: int = Path(..., description="战队 ID"),
current_user: User = Security(get_client_user),
):
team = await session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
if team.leader_id != current_user.id:
raise HTTPException(status_code=403, detail="You are not the team leader")
team_members = await session.exec(
select(TeamMember).where(TeamMember.team_id == team_id)
)
for member in team_members:
await session.delete(member)
await session.delete(team)
await session.commit()
class TeamQueryResp(BaseModel):
team: Team
members: list[UserResp]
@router.get("/team/{team_id}", name="查询战队", response_model=TeamQueryResp)
async def get_team(
session: Database,
team_id: int = Path(..., description="战队 ID"),
):
members = (
await session.exec(select(TeamMember).where(TeamMember.team_id == team_id))
).all()
return TeamQueryResp(
team=members[0].team,
members=[
await UserResp.from_db(m.user, session, include=BASE_INCLUDES)
for m in members
],
)
@router.post("/team/{team_id}/request", name="请求加入战队", status_code=204)
async def request_join_team(
session: Database,
team_id: int = Path(..., description="战队 ID"),
current_user: User = Security(get_client_user),
):
team = await session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
if (await current_user.awaitable_attrs.team_membership) is not None:
raise HTTPException(status_code=403, detail="You are already in a team")
if (
await session.exec(
select(exists()).where(
TeamRequest.team_id == team_id, TeamRequest.user_id == current_user.id
)
)
).first():
raise HTTPException(status_code=409, detail="Join request already exists")
team_request = TeamRequest(
user_id=current_user.id, team_id=team_id, requested_at=datetime.now(UTC)
)
session.add(team_request)
await session.commit()
await session.refresh(team_request)
await server.new_private_notification(TeamApplicationStore.init(team_request))
@router.post("/team/{team_id}/{user_id}/request", name="接受加入请求", status_code=204)
@router.delete(
"/team/{team_id}/{user_id}/request", name="拒绝加入请求", status_code=204
)
async def handle_request(
req: Request,
session: Database,
team_id: int = Path(..., description="战队 ID"),
user_id: int = Path(..., description="用户 ID"),
current_user: User = Security(get_client_user),
):
team = await session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
if team.leader_id != current_user.id:
raise HTTPException(status_code=403, detail="You are not the team leader")
team_request = (
await session.exec(
select(TeamRequest).where(
TeamRequest.team_id == team_id, TeamRequest.user_id == user_id
)
)
).first()
if not team_request:
raise HTTPException(status_code=404, detail="Join request not found")
user = await session.get(User, user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found")
if req.method == "POST":
if (
await session.exec(select(exists()).where(TeamMember.user_id == user_id))
).first():
raise HTTPException(
status_code=409, detail="User is already a member of the team"
)
session.add(
TeamMember(user_id=user_id, team_id=team_id, joined_at=datetime.now(UTC))
)
await server.new_private_notification(TeamApplicationAccept.init(team_request))
else:
await server.new_private_notification(TeamApplicationReject.init(team_request))
await session.delete(team_request)
await session.commit()
@router.delete("/team/{team_id}/{user_id}", name="踢出成员 / 退出战队", status_code=204)
async def kick_member(
session: Database,
team_id: int = Path(..., description="战队 ID"),
user_id: int = Path(..., description="用户 ID"),
current_user: User = Security(get_client_user),
):
team = await session.get(Team, team_id)
if not team:
raise HTTPException(status_code=404, detail="Team not found")
if team.leader_id != current_user.id and user_id != current_user.id:
raise HTTPException(status_code=403, detail="You are not the team leader")
team_member = (
await session.exec(
select(TeamMember).where(
TeamMember.team_id == team_id, TeamMember.user_id == user_id
)
)
).first()
if not team_member:
raise HTTPException(status_code=404, detail="User is not a member of the team")
if team.leader_id == current_user.id:
raise HTTPException(
status_code=403, detail="You cannot leave because you are the team leader"
)
await session.delete(team_member)
await session.commit()

22
app/utils.py
View File

@@ -1,6 +1,10 @@
from __future__ import annotations
from datetime import datetime
from io import BytesIO
from fastapi import HTTPException
from PIL import Image
def unix_timestamp_to_windows(timestamp: int) -> int:
@@ -126,6 +130,7 @@ def truncate(text: str, limit: int = 100, ellipsis: str = "...") -> str:
return text
def parse_user_agent(user_agent: str | None, max_length: int = 255) -> str | None:
"""
解析用户代理字符串，提取关键信息：设备、系统、浏览器
@@ -181,3 +186,20 @@ def parse_user_agent(user_agent: str | None, max_length: int = 255) -> str | Non
# 如果无法解析，则截断原始字符串
return truncate(user_agent, max_length - 3, "...")
def check_image(content: bytes, size: int, width: int, height: int) -> None:
if len(content) > size: # 10MB limit
raise HTTPException(status_code=400, detail="File size exceeds 10MB limit")
elif len(content) == 0:
raise HTTPException(status_code=400, detail="File cannot be empty")
try:
with Image.open(BytesIO(content)) as img:
if img.format not in ["PNG", "JPEG", "GIF"]:
raise HTTPException(status_code=400, detail="Invalid image format")
if img.size[0] > width or img.size[1] > height:
raise HTTPException(
status_code=400,
detail=f"Image size exceeds {width}x{height} pixels",
)
except Exception as e:
raise HTTPException(status_code=400, detail=f"Error processing image: {e}")

86
migrations/versions/65e7dc8d5905_team_add_team_request_table.py Normal file
View File

@@ -0,0 +1,86 @@
"""team: add team request table
Revision ID: 65e7dc8d5905
Revises: d103d442dc24
Create Date: 2025-08-22 03:47:57.870398
"""
from __future__ import annotations
from collections.abc import Sequence
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql
import sqlmodel
# revision identifiers, used by Alembic.
revision: str = "65e7dc8d5905"
down_revision: str | Sequence[str] | None = "d103d442dc24"
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! ###
op.create_table(
"team_requests",
sa.Column("user_id", sa.BigInteger(), nullable=False),
sa.Column("team_id", sa.Integer(), nullable=False),
sa.Column("requested_at", sa.DateTime(), nullable=True),
sa.ForeignKeyConstraint(
["team_id"],
["teams.id"],
),
sa.ForeignKeyConstraint(
["user_id"],
["lazer_users.id"],
),
sa.PrimaryKeyConstraint("user_id", "team_id"),
)
op.alter_column(
"team_members", "user_id", existing_type=mysql.BIGINT(), nullable=False
)
op.drop_index(op.f("ix_team_members_id"), table_name="team_members")
op.drop_column("team_members", "id")
op.add_column(
"teams",
sa.Column("cover_url", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
)
op.add_column("teams", sa.Column("leader_id", sa.BigInteger(), nullable=True))
op.alter_column(
"teams",
"flag_url",
existing_type=mysql.VARCHAR(length=500),
type_=sqlmodel.sql.sqltypes.AutoString(),
existing_nullable=True,
)
op.create_foreign_key(None, "teams", "lazer_users", ["leader_id"], ["id"])
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(None, "teams", type_="foreignkey")
op.alter_column(
"teams",
"flag_url",
existing_type=sqlmodel.sql.sqltypes.AutoString(),
type_=mysql.VARCHAR(length=500),
existing_nullable=True,
)
op.drop_column("teams", "leader_id")
op.drop_column("teams", "cover_url")
op.add_column(
"team_members",
sa.Column("id", mysql.INTEGER(), autoincrement=True, nullable=False),
)
op.create_index(op.f("ix_team_members_id"), "team_members", ["id"], unique=False)
op.alter_column(
"team_members", "user_id", existing_type=mysql.BIGINT(), nullable=True
)
op.drop_table("team_requests")
# ### end Alembic commands ###