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

refactor(project): use unified utcnow

Browse Source
This commit is contained in:
MingxuanGame
2025-08-22 11:27:45 +00:00
parent da66420eaa
commit 9b00dbda28
49 changed files with 201 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
app/database/achievement.py
View File

@@ -1,10 +1,11 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import TYPE_CHECKING
from app.config import settings
from app.models.achievement import MEDALS, Achievement
from app.models.model import UTCBaseModel
from app.models.notification import UserAchievementUnlock
from app.utils import utcnow
from .events import Event, EventType
@@ -28,7 +29,7 @@ if TYPE_CHECKING:
class UserAchievementBase(SQLModel, UTCBaseModel):
achievement_id: int
achieved_at: datetime = Field(default=datetime.now(UTC), sa_column=Column(DateTime(timezone=True)))
achieved_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime(timezone=True)))
class UserAchievement(UserAchievementBase, table=True):
@@ -56,7 +57,7 @@ async def process_achievements(session: AsyncSession, redis: Redis, score_id: in
).all()
not_achieved = {k: v for k, v in MEDALS.items() if k.id not in achieved}
result: list[Achievement] = []
now = datetime.now(UTC)
now = utcnow()
for k, v in not_achieved.items():
if await v(session, score, score.beatmap):
result.append(k)

3
app/database/auth.py
View File

@@ -3,6 +3,7 @@ import secrets
from typing import TYPE_CHECKING
from app.models.model import UTCBaseModel
from app.utils import utcnow
from sqlalchemy import Column, DateTime
from sqlmodel import (
@@ -30,7 +31,7 @@ class OAuthToken(UTCBaseModel, SQLModel, table=True):
token_type: str = Field(default="Bearer", max_length=20)
scope: str = Field(default="*", max_length=100)
expires_at: datetime = Field(sa_column=Column(DateTime))
created_at: datetime = Field(default_factory=datetime.utcnow, sa_column=Column(DateTime))
created_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
user: "User" = Relationship()

4
app/database/beatmap_playcounts.py
View File

@@ -1,7 +1,7 @@
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from app.database.events import Event, EventType
from app.utils import utcnow
from pydantic import BaseModel
from sqlalchemy.ext.asyncio import AsyncAttrs
@@ -71,7 +71,7 @@ async def process_beatmap_playcount(session: AsyncSession, user_id: int, beatmap
existing_playcount.playcount += 1
if existing_playcount.playcount % 100 == 0:
playcount_event = Event(
created_at=datetime.now(UTC),
created_at=utcnow(),
type=EventType.BEATMAP_PLAYCOUNT,
user_id=user_id,
)

7
app/database/chat.py
View File

@@ -1,9 +1,10 @@
from datetime import UTC, datetime
from datetime import datetime
from enum import Enum
from typing import Self
from app.database.lazer_user import RANKING_INCLUDES, User, UserResp
from app.models.model import UTCBaseModel
from app.utils import utcnow
from pydantic import BaseModel
from redis.asyncio import Redis
@@ -170,7 +171,7 @@ class ChatMessageBase(UTCBaseModel, SQLModel):
content: str = Field(sa_column=Column(VARCHAR(1000)))
message_id: int = Field(index=True, primary_key=True, default=None)
sender_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), index=True))
timestamp: datetime = Field(sa_column=Column(DateTime, index=True), default=datetime.now(UTC))
timestamp: datetime = Field(sa_column=Column(DateTime, index=True), default_factory=utcnow)
type: MessageType = Field(default=MessageType.PLAIN, index=True, exclude=True)
uuid: str | None = Field(default=None)
@@ -208,7 +209,7 @@ class SilenceUser(UTCBaseModel, SQLModel, table=True):
channel_id: int = Field(foreign_key="chat_channels.channel_id", index=True)
until: datetime | None = Field(sa_column=Column(DateTime, index=True), default=None)
reason: str | None = Field(default=None, sa_column=Column(VARCHAR(255), index=True))
banned_at: datetime = Field(sa_column=Column(DateTime, index=True), default=datetime.now(UTC))
banned_at: datetime = Field(sa_column=Column(DateTime, index=True), default_factory=utcnow)
class UserSilenceResp(SQLModel):

4
app/database/daily_challenge.py
View File

@@ -2,7 +2,7 @@ from datetime import UTC, datetime, timedelta
from typing import TYPE_CHECKING
from app.models.model import UTCBaseModel
from app.utils import are_adjacent_weeks
from app.utils import are_adjacent_weeks, utcnow
from sqlmodel import (
BigInteger,
@@ -79,7 +79,7 @@ async def process_daily_challenge_score(session: AsyncSession, user_id: int, roo
session.add(stats)
stats.playcount += 1
now = datetime.now(UTC)
now = utcnow()
if stats.last_update is None:
stats.daily_streak_best = 1
stats.daily_streak_current = 1

8
app/database/email_verification.py
View File

@@ -4,7 +4,9 @@
from __future__ import annotations
from datetime import UTC, datetime
from datetime import datetime
from app.utils import utcnow
from sqlalchemy import BigInteger, Column, ForeignKey
from sqlmodel import Field, SQLModel
@@ -19,7 +21,7 @@ class EmailVerification(SQLModel, table=True):
user_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), nullable=False, index=True))
email: str = Field(index=True)
verification_code: str = Field(max_length=8) # 8位验证码
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
created_at: datetime = Field(default_factory=utcnow)
expires_at: datetime = Field() # 验证码过期时间
is_used: bool = Field(default=False) # 是否已使用
used_at: datetime | None = Field(default=None)
@@ -39,7 +41,7 @@ class LoginSession(SQLModel, table=True):
user_agent: str | None = Field(default=None, max_length=250)
country_code: str | None = Field(default=None)
is_verified: bool = Field(default=False) # 是否已验证
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
created_at: datetime = Field(default_factory=lambda: utcnow())
verified_at: datetime | None = Field(default=None)
expires_at: datetime = Field() # 会话过期时间
is_new_location: bool = Field(default=False) # 是否新位置登录

3
app/database/events.py
View File

@@ -3,6 +3,7 @@ from enum import Enum
from typing import TYPE_CHECKING
from app.models.model import UTCBaseModel
from app.utils import utcnow
from pydantic import model_serializer
from sqlmodel import (
@@ -40,7 +41,7 @@ class EventType(str, Enum):
class Event(UTCBaseModel, SQLModel, table=True):
__tablename__: str = "user_events"
id: int = Field(default=None, primary_key=True)
created_at: datetime = Field(sa_column=Column(DateTime(timezone=True), default=datetime.now(UTC)))
created_at: datetime = Field(sa_column=Column(DateTime(timezone=True), default_factory=utcnow))
type: EventType
event_payload: dict = Field(exclude=True, default_factory=dict, sa_column=Column(JSON))
user_id: int | None = Field(

11
app/database/lazer_user.py
View File

@@ -1,4 +1,4 @@
from datetime import UTC, datetime, timedelta
from datetime import datetime, timedelta
import json
from typing import TYPE_CHECKING, NotRequired, TypedDict
@@ -6,6 +6,7 @@ from app.models.model import UTCBaseModel
from app.models.score import GameMode
from app.models.user import Country, Page
from app.path import STATIC_DIR
from app.utils import utcnow
from .achievement import UserAchievement, UserAchievementResp
from .beatmap_playcounts import BeatmapPlaycounts
@@ -75,7 +76,7 @@ class UserBase(UTCBaseModel, SQLModel):
is_active: bool = True
is_bot: bool = False
is_supporter: bool = False
last_visit: datetime | None = Field(default=datetime.now(UTC), sa_column=Column(DateTime(timezone=True)))
last_visit: datetime | None = Field(default_factory=utcnow, sa_column=Column(DateTime(timezone=True)))
pm_friends_only: bool = False
profile_colour: str | None = None
username: str = Field(max_length=32, unique=True, index=True)
@@ -99,7 +100,7 @@ class UserBase(UTCBaseModel, SQLModel):
discord: str | None = None
has_supported: bool = False
interests: str | None = None
join_date: datetime = Field(default=datetime.now(UTC))
join_date: datetime = Field(default_factory=utcnow)
location: str | None = None
max_blocks: int = 50
max_friends: int = 500
@@ -408,7 +409,7 @@ class UserResp(UserBase):
Score.user_id == obj.id,
Score.gamemode == ruleset,
col(Score.passed).is_(True),
Score.ended_at > datetime.now(UTC) - timedelta(hours=24),
Score.ended_at > utcnow() - timedelta(hours=24),
)
)
).one()
@@ -437,7 +438,7 @@ class UserResp(UserBase):
select(LoginSession).where(
LoginSession.user_id == obj.id,
col(LoginSession.is_verified).is_(False),
LoginSession.expires_at > datetime.now(UTC),
LoginSession.expires_at > utcnow(),
)
)
).first()

7
app/database/multiplayer_event.py
View File

@@ -1,7 +1,8 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import Any
from app.models.model import UTCBaseModel
from app.utils import utcnow
from sqlmodel import (
JSON,
@@ -24,7 +25,7 @@ class MultiplayerEventBase(SQLModel, UTCBaseModel):
sa_column=Column(
DateTime(timezone=True),
),
default=datetime.now(UTC),
default_factory=utcnow,
)
event_type: str = Field(index=True)
@@ -40,7 +41,7 @@ class MultiplayerEvent(MultiplayerEventBase, table=True):
sa_column=Column(
DateTime(timezone=True),
),
default=datetime.now(UTC),
default_factory=utcnow,
)
event_detail: dict[str, Any] | None = Field(
sa_column=Column(JSON),

5
app/database/notification.py
View File

@@ -1,7 +1,8 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import Any
from app.models.notification import NotificationDetail, NotificationName
from app.utils import utcnow
from sqlmodel import (
JSON,
@@ -54,7 +55,7 @@ async def insert_notification(session: AsyncSession, detail: NotificationDetail)
object_id=detail.object_id,
source_user_id=detail.source_user_id,
details=detail.model_dump(),
created_at=datetime.now(UTC),
created_at=utcnow(),
)
session.add(notification)
await session.commit()

6
app/database/password_reset.py
View File

@@ -4,7 +4,9 @@
from __future__ import annotations
from datetime import UTC, datetime
from datetime import datetime
from app.utils import utcnow
from sqlalchemy import BigInteger, Column, ForeignKey
from sqlmodel import Field, SQLModel
@@ -19,7 +21,7 @@ class PasswordReset(SQLModel, table=True):
user_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), nullable=False, index=True))
email: str = Field(index=True)
reset_code: str = Field(max_length=8) # 8位重置验证码
created_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
created_at: datetime = Field(default_factory=utcnow)
expires_at: datetime = Field() # 验证码过期时间
is_used: bool = Field(default=False) # 是否已使用
used_at: datetime | None = Field(default=None)

7
app/database/rank_history.py
View File

@@ -1,11 +1,10 @@
from datetime import (
UTC,
date as dt,
datetime,
)
from typing import TYPE_CHECKING, Optional
from app.models.score import GameMode
from app.utils import utcnow
from pydantic import BaseModel
from sqlmodel import (
@@ -33,7 +32,7 @@ class RankHistory(SQLModel, table=True):
mode: GameMode
rank: int
date: dt = Field(
default_factory=lambda: datetime.now(UTC).date(),
default_factory=lambda: utcnow().date(),
sa_column=Column(Date, index=True),
)
@@ -48,7 +47,7 @@ class RankTop(SQLModel, table=True):
mode: GameMode
rank: int
date: dt = Field(
default_factory=lambda: datetime.now(UTC).date(),
default_factory=lambda: utcnow().date(),
sa_column=Column(Date, index=True),
)

5
app/database/room.py
View File

@@ -1,4 +1,4 @@
from datetime import UTC, datetime
from datetime import datetime
from app.database.playlist_attempts import PlaylistAggregateScore
from app.database.room_participated_user import RoomParticipatedUser
@@ -12,6 +12,7 @@ from app.models.room import (
RoomPlaylistItemStats,
RoomStatus,
)
from app.utils import utcnow
from .lazer_user import User, UserResp
from .playlists import Playlist, PlaylistResp
@@ -39,7 +40,7 @@ class RoomBase(SQLModel, UTCBaseModel):
sa_column=Column(
DateTime(timezone=True),
),
default=datetime.now(UTC),
default_factory=utcnow,
)
ends_at: datetime | None = Field(
sa_column=Column(

6
app/database/room_participated_user.py
View File

@@ -1,6 +1,8 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import TYPE_CHECKING
from app.utils import utcnow
from sqlalchemy.ext.asyncio import AsyncAttrs
from sqlmodel import (
BigInteger,
@@ -25,7 +27,7 @@ class RoomParticipatedUser(AsyncAttrs, SQLModel, table=True):
user_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id"), nullable=False))
joined_at: datetime = Field(
sa_column=Column(DateTime(timezone=True), nullable=False),
default=datetime.now(UTC),
default_factory=utcnow,
)
left_at: datetime | None = Field(sa_column=Column(DateTime(timezone=True), nullable=True), default=None)

7
app/database/score.py
View File

@@ -1,5 +1,5 @@
from collections.abc import Sequence
from datetime import UTC, date, datetime
from datetime import date, datetime
import json
import math
import sys
@@ -31,6 +31,7 @@ from app.models.score import (
ScoreStatistics,
SoloScoreSubmissionInfo,
)
from app.utils import utcnow
from .beatmap import Beatmap, BeatmapResp
from .beatmap_playcounts import process_beatmap_playcount
@@ -733,7 +734,7 @@ async def process_user(
if i < score_range and displaced_position > score_range and displaced_position is not None:
# Create rank lost event for the displaced user
rank_lost_event = Event(
created_at=datetime.now(UTC),
created_at=utcnow(),
type=EventType.RANK_LOST,
user_id=displaced_score.user_id,
)
@@ -843,7 +844,7 @@ async def process_score(
total_score=info.total_score,
total_score_without_mods=info.total_score_without_mods,
beatmap_id=beatmap_id,
ended_at=datetime.now(UTC),
ended_at=utcnow(),
gamemode=gamemode,
started_at=score_token.created_at,
user_id=user.id,

5
app/database/score_token.py
View File

@@ -2,6 +2,7 @@ from datetime import datetime
from app.models.model import UTCBaseModel
from app.models.score import GameMode
from app.utils import utcnow
from .beatmap import Beatmap
from .lazer_user import User
@@ -15,8 +16,8 @@ class ScoreTokenBase(SQLModel, UTCBaseModel):
score_id: int | None = Field(sa_column=Column(BigInteger), default=None)
ruleset_id: GameMode
playlist_item_id: int | None = Field(default=None) # playlist
created_at: datetime = Field(default_factory=datetime.utcnow, sa_column=Column(DateTime))
updated_at: datetime = Field(default_factory=datetime.utcnow, sa_column=Column(DateTime))
created_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
updated_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
class ScoreToken(ScoreTokenBase, table=True):

7
app/database/statistics.py
View File

@@ -1,8 +1,9 @@
from datetime import UTC, datetime, timedelta
from datetime import timedelta
import math
from typing import TYPE_CHECKING
from app.models.score import GameMode
from app.utils import utcnow
from .rank_history import RankHistory
@@ -134,7 +135,7 @@ class UserStatisticsResp(UserStatisticsBase):
rank_best = (
await session.exec(
select(func.max(RankHistory.rank)).where(
RankHistory.date > datetime.now(UTC) - timedelta(days=30),
RankHistory.date > utcnow() - timedelta(days=30),
RankHistory.user_id == obj.user_id,
)
)
@@ -171,7 +172,7 @@ async def get_rank(session: AsyncSession, statistics: UserStatistics, country: s
return None
if country is None:
today = datetime.now(UTC).date()
today = utcnow().date()
rank_history = (
await session.exec(
select(RankHistory).where(

9
app/database/team.py
View File

@@ -1,7 +1,8 @@
from datetime import UTC, datetime
from datetime import datetime
from typing import TYPE_CHECKING
from app.models.model import UTCBaseModel
from app.utils import utcnow
from sqlalchemy import Column, DateTime
from sqlmodel import BigInteger, Field, ForeignKey, Relationship, SQLModel
@@ -18,7 +19,7 @@ class Team(SQLModel, UTCBaseModel, table=True):
short_name: str = Field(max_length=10)
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))
created_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
leader_id: int = Field(sa_column=Column(BigInteger, ForeignKey("lazer_users.id")))
leader: "User" = Relationship()
@@ -30,7 +31,7 @@ class TeamMember(SQLModel, UTCBaseModel, table=True):
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))
joined_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
user: "User" = Relationship(back_populates="team_membership", sa_relationship_kwargs={"lazy": "joined"})
team: "Team" = Relationship(back_populates="members", sa_relationship_kwargs={"lazy": "joined"})
@@ -41,7 +42,7 @@ class TeamRequest(SQLModel, UTCBaseModel, table=True):
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))
requested_at: datetime = Field(default_factory=utcnow, sa_column=Column(DateTime))
user: "User" = Relationship(sa_relationship_kwargs={"lazy": "joined"})
team: "Team" = Relationship(sa_relationship_kwargs={"lazy": "joined"})

5
app/database/user_account_history.py
View File

@@ -1,7 +1,8 @@
from datetime import UTC, datetime
from datetime import datetime
from enum import Enum
from app.models.model import UTCBaseModel
from app.utils import utcnow
from sqlmodel import BigInteger, Column, Field, ForeignKey, Integer, SQLModel
@@ -17,7 +18,7 @@ class UserAccountHistoryBase(SQLModel, UTCBaseModel):
description: str | None = None
length: int
permanent: bool = False
timestamp: datetime = Field(default=datetime.now(UTC))
timestamp: datetime = Field(default_factory=utcnow)
type: UserAccountHistoryType

4
app/database/user_login_log.py
View File

@@ -4,6 +4,8 @@ User Login Log Database Model
from datetime import datetime
from app.utils import utcnow
from sqlmodel import Field, SQLModel
@@ -16,7 +18,7 @@ class UserLoginLog(SQLModel, table=True):
user_id: int = Field(index=True, description="User ID")
ip_address: str = Field(max_length=45, index=True, description="IP address (supports IPv4 and IPv6)")
user_agent: str | None = Field(default=None, max_length=500, description="User agent information")
login_time: datetime = Field(default_factory=datetime.utcnow, description="Login time")
login_time: datetime = Field(default_factory=utcnow, description="Login time")
# GeoIP information
country_code: str | None = Field(default=None, max_length=2, description="Country code")