together_app
/
together_backend
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

implemented apply friend

main
htylight 2023-07-27 18:19:21 +08:00
parent 28e72d7467
commit f512089fd9
18 changed files with 587 additions and 121 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
migrations/env.py
View File

@ -46,6 +46,8 @@ def run_migrations_offline() -> None:
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
compare_type=True,
compare_server_default=True,
)
with context.begin_transaction():

22
migrations/versions/3a1f57242769_create_user_account_user_profile_and_.py → migrations/versions/c2ce366349b6_create_table_user_account_user_profile_.py 100755 → 100644
View File

@ -1,8 +1,8 @@
"""create user_account, user_profile and contact table
"""create table user_account, user_profile, contact and apply
Revision ID: 3a1f57242769
Revision ID: c2ce366349b6
Revises:
Create Date: 2023-07-07 21:55:20.913012
Create Date: 2023-07-22 20:42:45.268914
"""
from alembic import op
@ -10,7 +10,7 @@ import sqlalchemy as sa
from sqlalchemy.dialects import postgresql
# revision identifiers, used by Alembic.
revision = '3a1f57242769'
revision = 'c2ce366349b6'
down_revision = None
branch_labels = None
depends_on = None
@ -18,6 +18,17 @@ depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table('apply',
sa.Column('id', sa.Integer(), nullable=False),
sa.Column('relation', sa.Integer(), nullable=False),
sa.Column('applicant', sa.String(length=26), nullable=False),
sa.Column('recipient', sa.String(length=26), nullable=False),
sa.Column('inviter', sa.String(length=26), nullable=True),
sa.Column('hello', sa.String(), nullable=False),
sa.Column('setting', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column('created_at', sa.DateTime(), nullable=False),
sa.PrimaryKeyConstraint('id')
)
op.create_table('user_account',
sa.Column('id', sa.String(length=26), nullable=False),
sa.Column('username', sa.String(length=20), nullable=False),
@ -46,7 +57,7 @@ def upgrade() -> None:
sa.Column('location', sa.String(), nullable=True),
sa.Column('status', sa.String(), nullable=True),
sa.Column('sign', sa.String(), nullable=True),
sa.Column('avatars', sa.String(), nullable=True),
sa.Column('avatar', sa.String(), nullable=True),
sa.Column('user_id', sa.String(length=26), nullable=False),
sa.ForeignKeyConstraint(['user_id'], ['user_account.id'], ondelete='CASCADE'),
sa.PrimaryKeyConstraint('id')
@ -59,4 +70,5 @@ def downgrade() -> None:
op.drop_table('user_profile')
op.drop_table('contact')
op.drop_table('user_account')
op.drop_table('apply')
# ### end Alembic commands ###

30
migrations/versions/d0c7f4dd4894_change_inviter_column_to_grou_chat_id_.py 100644
View File

@ -0,0 +1,30 @@
"""change inviter column to grou_chat_id of apply table
Revision ID: d0c7f4dd4894
Revises: c2ce366349b6
Create Date: 2023-07-23 18:32:31.233737
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'd0c7f4dd4894'
down_revision = 'c2ce366349b6'
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('apply', sa.Column('group_chat_id', sa.String(length=26), nullable=True))
op.drop_column('apply', 'inviter')
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('apply', sa.Column('inviter', sa.VARCHAR(length=26), autoincrement=False, nullable=True))
op.drop_column('apply', 'group_chat_id')
# ### end Alembic commands ###

71
src/crud/apply_crud.py 100644
View File

@ -0,0 +1,71 @@
from typing import Literal, Tuple, Union
from datetime import date
from sqlalchemy import select, delete, update, insert
from sqlalchemy import ScalarResult
from sqlalchemy.exc import NoResultFound
from ..database.db import async_session
from ..database.models import Apply
async def insert_apply(
relation: int,
applicant: str,
recipient: str,
group_chat_id: str | None,
hello: str,
setting: dict[str, str],
):
session = async_session()
res: ScalarResult[Apply] = await session.scalars(
select(Apply).where(Apply.recipient == recipient, Apply.applicant == applicant)
)
try:
apply = res.one()
apply.hello = hello
apply.setting = setting
session.add(apply)
except NoResultFound:
apply = Apply(
relation=relation,
applicant=applicant,
recipient=recipient,
group_chat_id=group_chat_id,
hello=hello,
setting=setting,
)
session.add(apply)
finally:
await session.commit()
await session.close()
async def select_apply_all(recipient: str) -> list[Apply]:
session = async_session()
res: ScalarResult[Apply] = await session.scalars(
select(Apply)
.where(Apply.recipient == recipient)
.order_by(Apply.created_at.desc())
)
try:
return list(res.all())
except NoResultFound:
return []
finally:
await session.close()
async def delete_apply(relation: int, applicant: str, recipient: str):
session = async_session()
await session.execute(
delete(Apply).where(
Apply.relation == relation,
Apply.applicant == applicant,
Apply.recipient == recipient,
)
)
await session.commit()
await session.close()

43
src/crud/multitable_crud.py 100644
View File

@ -0,0 +1,43 @@
from sqlalchemy import select, delete, update
from sqlalchemy import Result, ScalarResult, or_
from sqlalchemy.orm.attributes import flag_modified
from ..database.db import async_session
from ..database.models import *
async def insert_contact_friend(
relation: int,
applicant: str,
recipient: str,
applicant_setting: dict,
recipient_setting: dict,
):
session = async_session()
try:
await session.execute(
delete(Apply).where(
Apply.recipient == recipient,
Apply.applicant == applicant,
Apply.relation == relation,
)
)
res: ScalarResult[Contact] = await session.scalars(
select(Contact).where(
or_(Contact.user_id == applicant, Contact.user_id == recipient)
)
)
for contact in res.all():
if contact.user_id == recipient:
contact.friends[applicant] = recipient_setting
flag_modified(contact, "friends")
else:
contact.friends[recipient] = applicant_setting
flag_modified(contact, "friends")
session.add_all(res.all())
await session.commit()
except Exception:
raise Exception
finally:
await session.close()

153
src/crud/user_crud.py
View File

@ -1,40 +1,22 @@
from typing import Literal, Tuple
from typing import Literal, Tuple, Union
from datetime import date
from sqlalchemy import select, update
from sqlalchemy import select, update, Row, Result, Sequence
from sqlalchemy import ScalarResult
import ulid
from sqlalchemy.exc import NoResultFound
from ..database.db import async_session
from ..database.models import UserAccount, UserProfile, Contact
async def select_user_by(condition: Literal['email', 'username', 'id'], value: str) -> Tuple[bool, UserAccount]:
session = async_session()
res: ScalarResult[UserAccount] = ScalarResult[UserAccount]
match condition:
case 'email':
res = await session.scalars(select(UserAccount).where(UserAccount.email == value))
case 'username':
res = await session.scalars(select(UserAccount).where(UserAccount.username == value))
case 'id':
res = await session.scalars(select(UserAccount).where(UserAccount.id == value))
user = res.first()
return (True, user) if user else (False, None)
async def insert_user(username: str, password: str, email: str):
session = async_session()
id = ulid.new().str
user = UserAccount(id=id, username=username, password=password, email=email)
profile = UserProfile(nickname=username)
contact = Contact(
friends={id: {'friendRemark': None, 'friendGroup': '我的好友'}}, friend_groups=['我的好友'],
group_chats={}
)
contact = Contact(friends={id: {}}, friend_groups=["我的好友"], group_chats={})
user.profile = profile
user.contact = contact
session.add(user)
@ -42,20 +24,78 @@ async def insert_user(username: str, password: str, email: str):
await session.close()
async def select_account_by(
condition: Literal["email", "username", "id"], value: str
) -> Tuple[bool, UserAccount]:
session = async_session()
res: ScalarResult[UserAccount] = ScalarResult[UserAccount]
match condition:
case "email":
res = await session.scalars(
select(UserAccount).where(UserAccount.email == value)
)
case "username":
res = await session.scalars(
select(UserAccount).where(UserAccount.username == value)
)
case "id":
res = await session.scalars(
select(UserAccount).where(UserAccount.id == value)
)
user = res.first()
return (True, user) if user else (False, None)
async def update_account(
account: Literal["username", "email", "password"], id: str, value: str
):
session = async_session()
match account:
case "username":
await session.execute(
update(UserAccount).where(UserAccount.id == id).values(username=value)
)
case "email":
await session.execute(
update(UserAccount).where(UserAccount.id == id).values(email=value)
)
case "password":
await session.execute(
update(UserAccount).where(UserAccount.id == id).values(password=value)
)
await session.commit()
await session.close()
async def select_profile(id: str) -> UserProfile:
session = async_session()
res: ScalarResult[UserProfile] = await session.scalars(select(UserProfile).where(UserProfile.user_id == id))
res: ScalarResult[UserProfile] = await session.scalars(
select(UserProfile).where(UserProfile.user_id == id)
)
res: UserProfile = res.first()
await session.close()
return res
async def update_profile_basic(id: str, nickname: str, location: str, birthday: str, gender: str, ):
async def update_profile_basic(
id: str,
nickname: str,
location: str,
birthday: str,
gender: str,
):
session = async_session()
await session.execute(
update(UserProfile)
.where(UserProfile.user_id == id)
.values(nickname=nickname, location=location, birthday=date.fromisoformat(birthday), gender=gender)
.values(
nickname=nickname,
location=location,
birthday=date.fromisoformat(birthday),
gender=gender,
)
)
await session.commit()
await session.close()
@ -64,9 +104,7 @@ async def update_profile_basic(id: str, nickname: str, location: str, birthday:
async def update_profile_sign(id: str, sign: str):
session = async_session()
await session.execute(
update(UserProfile)
.where(UserProfile.user_id == id)
.values(sign=sign)
update(UserProfile).where(UserProfile.user_id == id).values(sign=sign)
)
await session.commit()
await session.close()
@ -75,34 +113,49 @@ async def update_profile_sign(id: str, sign: str):
async def update_profile_avatar(id: str, avatar_name: str):
session = async_session()
await session.execute(
update(UserProfile)
.where(UserProfile.user_id == id)
.values(avatar=avatar_name)
update(UserProfile).where(UserProfile.user_id == id).values(avatar=avatar_name)
)
await session.commit()
await session.close()
async def update_account(account: Literal['username', 'email', 'password'], id: str, value: str):
async def select_account_profile(
condition: Literal["username", "email"],
value: str,
) -> Tuple[bool, Union[Tuple[UserAccount, UserProfile], None]]:
session = async_session()
match account:
case 'username':
await session.execute(
update(UserAccount)
.where(UserAccount.id == id)
.values(username=value)
res: ScalarResult[Tuple[UserAccount, UserProfile]]
match condition:
case "username":
# must use join otherwise the result will be Cartesian product
res = await session.execute(
select(UserAccount, UserProfile)
.join(UserAccount.profile)
.where(UserAccount.username == value)
)
case 'email':
await session.execute(
update(UserAccount)
.where(UserAccount.id == id)
.values(email=value)
case "email":
res = await session.execute(
select(UserAccount, UserProfile)
.join(UserAccount.profile)
.where(UserAccount.email == value)
)
case 'password':
await session.execute(
update(UserAccount)
.where(UserAccount.id == id)
.values(password=value)
)
await session.commit()
await session.close()
try:
return True, res.one()
except NoResultFound:
return False, None
async def select_multiuser_info(
applicant_ids: list,
) -> Sequence[Row[Tuple[UserAccount, UserProfile]]]:
session = async_session()
res: Result = await session.execute(
select(UserAccount, UserProfile)
.join(UserAccount.profile)
.where(UserAccount.id.in_(applicant_ids))
)
await session.close()
return res.all()

11
src/database/json_typeddict.py 100644
View File

@ -0,0 +1,11 @@
from typing import TypedDict
class FriendSetting(TypedDict):
friendRemark: str | None
friendGroup: str
class GroupChatSetting(TypedDict):
groupChatRemark: str | None
myRemark: str | None

101
src/database/models.py
View File

@ -4,9 +4,11 @@ from enum import StrEnum
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import DeclarativeBase, relationship
from sqlalchemy.orm import mapped_column, Mapped
from sqlalchemy import String, Integer, Enum, DateTime, ARRAY, Date
from sqlalchemy import String, Integer, Enum, ARRAY, Date
from sqlalchemy import ForeignKey
from .json_typeddict import *
class Base(DeclarativeBase):
pass
@ -20,28 +22,30 @@ class UserAccount(Base):
email: Mapped[str] = mapped_column(String, unique=True)
password: Mapped[str] = mapped_column(String(60))
updated_at: Mapped[datetime] = mapped_column(default=datetime.now)
created_at: Mapped[datetime] = mapped_column(default=datetime.now, onupdate=datetime.now)
profile: Mapped['UserProfile'] = relationship(back_populates='user')
contact: Mapped['Contact'] = relationship(back_populates='user')
created_at: Mapped[datetime] = mapped_column(
default=datetime.now, onupdate=datetime.now
)
profile: Mapped["UserProfile"] = relationship(back_populates="user")
contact: Mapped["Contact"] = relationship(back_populates="user")
def __repr__(self):
return f'UserAccount(username={self.username}, email={self.email})'
return f"UserAccount(username={self.username}, email={self.email})"
def to_dict(self) -> dict:
return {
'id': self.id,
'username': self.username,
'email': self.email,
"id": self.id,
"username": self.username,
"email": self.email,
}
class Gender(StrEnum):
man = 'man'
woman = 'woman'
man = "man"
woman = "woman"
class UserProfile(Base):
__tablename__ = 'user_profile'
__tablename__ = "user_profile"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
nickname: Mapped[str] = mapped_column(String)
@ -51,31 +55,72 @@ class UserProfile(Base):
status: Mapped[str] = mapped_column(String, nullable=True)
sign: Mapped[str] = mapped_column(String, nullable=True)
avatar: Mapped[str] = mapped_column(String, nullable=True)
user_id: Mapped[str] = mapped_column(ForeignKey('user_account.id', ondelete='CASCADE'))
user: Mapped['UserAccount'] = relationship(back_populates='profile')
user_id: Mapped[str] = mapped_column(
ForeignKey("user_account.id", ondelete="CASCADE")
)
user: Mapped["UserAccount"] = relationship(back_populates="profile")
def __repr__(self):
return f'UserProfile(user={self.user_id},' \
f'nickname={self.nickname})'
return f"UserProfile(user={self.user_id}," f"nickname={self.nickname})"
def to_dict_all(self):
def to_dict(self):
return {
'nickname': self.nickname,
'gender': self.gender,
'birthday': self.birthday.isoformat(),
'location': self.location,
'status': self.status,
'sign': self.sign,
'avatar': self.avatar,
"nickname": self.nickname,
"gender": self.gender,
"birthday": self.birthday and self.birthday.isoformat(),
"location": self.location,
"status": self.status,
"sign": self.sign,
"avatar": self.avatar,
}
class Contact(Base):
__tablename__ = 'contact'
__tablename__ = "contact"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
friends: Mapped[dict] = mapped_column(JSONB, nullable=True)
friends: Mapped[dict[str, FriendSetting]] = mapped_column(JSONB, nullable=True)
friend_groups: Mapped[list[str]] = mapped_column(ARRAY(String), nullable=True)
group_chats: Mapped[dict] = mapped_column(JSONB, nullable=True)
user_id: Mapped[str] = mapped_column(ForeignKey('user_account.id', ondelete='CASCADE'))
user: Mapped['UserAccount'] = relationship(back_populates='contact')
group_chats: Mapped[dict[str, GroupChatSetting]] = mapped_column(
JSONB, nullable=True
)
user_id: Mapped[str] = mapped_column(
ForeignKey("user_account.id", ondelete="CASCADE")
)
user: Mapped["UserAccount"] = relationship(back_populates="contact")
def __repr__(self):
return (
f"Contact("
f"user={self.user_id}, "
f"friends={self.friends}, "
f"friend_group={self.friend_groups}, "
f"group_chats={self.group_chats}"
f")"
)
class Apply(Base):
__tablename__ = "apply"
id: Mapped[int] = mapped_column(Integer, primary_key=True)
relation: Mapped[int] = mapped_column(Integer)
applicant: Mapped[str] = mapped_column(String(26))
recipient: Mapped[str] = mapped_column(String(26))
group_chat_id: Mapped[str] = mapped_column(String(26), nullable=True)
hello: Mapped[str] = mapped_column(String)
setting: Mapped[dict] = mapped_column(JSONB)
created_at: Mapped[datetime] = mapped_column(
default=datetime.now, onupdate=datetime.now
)
def to_dict(self):
return {
"relation": self.relation,
"applicant": self.applicant,
"recipient": self.recipient,
"groupChatId": self.group_chat_id,
"hello": self.hello,
"setting": self.setting,
"createdAt": self.created_at.strftime("%y-%m-%d %H:%M:%S"),
}

19
src/main.py
View File

@ -1,4 +1,5 @@
from fastapi import FastAPI, Depends
from starlette.responses import FileResponse
from .dependencies import verify_token
from .utils.email_code import smtp
@ -6,21 +7,33 @@ from .routers.signin import router as signin_router
from .routers.signup import router as signup_router
from .routers.user_profile import router as user_profile_router
from .routers.user_account import router as user_account_router
from .routers.search import router as search_router
from .routers.apply import router as apply_router
from .utils.static_file import create_zip_file
app = FastAPI()
app.include_router(signup_router)
app.include_router(signin_router)
app.include_router(user_profile_router, dependencies=[Depends(verify_token)])
app.include_router(user_account_router, dependencies=[Depends(verify_token)])
app.include_router(search_router, dependencies=[Depends(verify_token)])
app.include_router(apply_router, dependencies=[Depends(verify_token)])
@app.on_event('shutdown')
@app.on_event("shutdown")
def close_smtp():
smtp.close()
@app.get('/')
@app.get("/")
async def main():
return {'code': 10000, 'msg': 'hello world'}
return {"code": 10000, "msg": "hello world"}
@app.get("/zipfile")
async def get_zipfile():
file = create_zip_file(
["luhptjjk1688921163.png", "pnjdvldw1688921358.png"], "avatars"
)
return FileResponse(file)

5
src/response_models/search_response.py 100644
View File

@ -0,0 +1,5 @@
from typing import Optional
from pydantic import BaseModel
from .base import BaseResponseModel

17
src/response_models/user_response.py
View File

@ -38,5 +38,22 @@ class UserProfileResponse(BaseResponseModel):
data: _UserProfile
class _UserAccountProfile(BaseModel):
id: str
username: str
email: str
nickname: str
gender: str | None
birthday: str | None
location: str | None
status: str | None
sign: str | None
avatar: str | None
class UserAccountProfileResponse(BaseResponseModel):
data: _UserAccountProfile
class UserAvatarResponse(BaseResponseModel):
data: str

93
src/routers/apply.py 100644
View File

@ -0,0 +1,93 @@
from fastapi import APIRouter, Query
from fastapi.responses import FileResponse
from pydantic import BaseModel
from src.crud import apply_crud, user_crud, multitable_crud
from ..response_models.base import BaseResponseModel
from ..utils.static_file import create_zip_file
router = APIRouter(prefix="/apply", tags=["apply"])
class ApplyInfo(BaseModel):
relation: int
applicant: str
recipient: str
group_chat_id: str | None = None
hello: str
setting: dict[str, str] = {}
class AcceptInfo(BaseModel):
relation: int
applicant: str
recipient: str
group_chat_id: str | None = None
applicant_setting: dict
recipient_setting: dict
class RefuseInfo(BaseModel):
relation: int
applicant: str
recipient: str
@router.post("/friend", response_model=BaseResponseModel)
async def apply_friend(apply_info: ApplyInfo):
await apply_crud.insert_apply(**apply_info.model_dump())
return {"code": 10600, "msg": "Apply Friend Successfully"}
@router.get("/list")
async def get_apply_list(recipient: str):
res = await apply_crud.select_apply_all(recipient)
if not res:
return {"code": 10601, "msg": "The Apply List is Empty"}
else:
data = []
for apply in res:
data.append(apply.to_dict())
print(data)
return {
"code": 10600,
"msg": "Get All Apply List Successfully",
"data": data,
}
@router.get("/applicant_profiles")
async def get_applicant_profiles(applicant_ids: list[str] = Query(default=None)):
res = await user_crud.select_multiuser_info(applicant_ids)
applicant_profiles = {}
for applicant in res:
applicant_profiles[applicant[0].id] = applicant[0].to_dict()
applicant_profiles[applicant[0].id].update(applicant[1].to_dict())
print(applicant_profiles)
return {
"code": 10600,
"msg": "Get Applicant Information Successfully",
"data": applicant_profiles,
}
@router.get("/applicant_avatars")
async def download_applicant_avatars(avatars: list[str] = Query(default=None)):
file_path = create_zip_file(avatars, "avatars")
return FileResponse(file_path)
@router.post("/accept")
async def accept_apply(accept_info: AcceptInfo):
try:
await multitable_crud.insert_contact_friend(**accept_info.model_dump())
return {"code": 10600, "msg": "Add Friend Successfully"}
except Exception as e:
print(f"接受添加好友请求出错....: {e}")
return {"code": 10601, "msg": "Something Went Wrong On the Server"}
@router.post("/refuse")
async def refuse_apply(refuse_info: RefuseInfo):
await apply_crud.delete_apply(**refuse_info.model_dump())
return {"code": 10600, "msg": "Refuse Apply Successfully"}

22
src/routers/search.py 100644
View File

@ -0,0 +1,22 @@
from typing import Literal
from fastapi import APIRouter
from ..crud import user_crud
from ..response_models.user_response import UserAccountProfileResponse
router = APIRouter(prefix="/search", tags=["search"])
@router.get("/friend", response_model=UserAccountProfileResponse)
async def search_contact_by(condition: Literal["username", "email"], value: str):
is_existence, res = await user_crud.select_account_profile(condition, value)
if not is_existence:
return {"code": 10501, "msg": "The User Does Not Exist"}
data = {}
data.update(res[0].to_dict())
data.update(res[1].to_dict())
return {"code": 10500, "msg": "Search Successfully", "data": data}

6
src/routers/signin.py
View File

@ -5,7 +5,7 @@ from pydantic import BaseModel
from jose import ExpiredSignatureError, JWTError
from ..crud.user_crud import select_user_by
from ..crud.user_crud import select_account_by
from ..utils.password import verify_password
from ..utils.token_handler import create_signin_token, oauth2_scheme, verify_signin_token
from ..response_models.user_response import UserAccountResponse, TokenCreationResponse, TokenSigninResponse
@ -22,7 +22,7 @@ class TokenPayload(BaseModel):
async def signin_by_username(form_data: OAuth2PasswordRequestForm = Depends()):
username = form_data.username
password = form_data.password
is_existence, user = await select_user_by('username', username)
is_existence, user = await select_account_by('username', username)
if not is_existence:
return {'code': 10201, 'msg': 'Username or Password Is Incorrect'}
@ -45,7 +45,7 @@ async def create_token(token_payload: TokenPayload):
async def signin_by_token(token: str = Depends(oauth2_scheme)):
try:
new_token, id = verify_signin_token(token)
_, user = await select_user_by('id', id)
_, user = await select_account_by('id', id)
if new_token:
return {'code': 10200, 'msg': 'Sign in Successfully', 'data': user.to_dict(), 'token': new_token}
else:

2
src/routers/signup.py
View File

@ -19,7 +19,7 @@ class SignUpAccount(BaseModel):
@router.get('/has_existed')
async def has_account_existed(condition: Literal['username', 'email'], value: str):
(res, _) = await user_crud.select_user_by(condition, value)
(res, _) = await user_crud.select_account_by(condition, value)
if res:
return {'code': 10101, 'msg': f'{condition.capitalize()} Has Existed'}
else:

15
src/routers/user_account.py
View File

@ -4,6 +4,7 @@ from pydantic import BaseModel
from ..crud import user_crud
from ..utils import password
from ..utils.email_code import send_email, has_code, verify_code
from ..response_models.base import BaseResponseModel
router = APIRouter(prefix='/user_account', tags=['user_account'])
@ -16,9 +17,9 @@ class ChangedAccount(BaseModel):
code: str | None = None
@router.post('/change/username')
@router.post('/change/username', response_model=BaseResponseModel)
async def change_username(changed_account: ChangedAccount):
is_existed, user = await user_crud.select_user_by('username', changed_account.username)
is_existed, user = await user_crud.select_account_by('username', changed_account.username)
if is_existed:
return {'code': 10401, 'msg': f'This Username ({changed_account.username}) Has Been Used'}
@ -27,9 +28,9 @@ async def change_username(changed_account: ChangedAccount):
return {'code': 10400, 'msg': 'Update Username Successfully'}
@router.get('/get/email_code')
@router.get('/get/email_code', response_model=BaseResponseModel)
async def get_change_email_code(email: str, background_tasks: BackgroundTasks):
is_existed, _ = await user_crud.select_user_by('email', email)
is_existed, _ = await user_crud.select_account_by('email', email)
if is_existed:
return {'code': 10401, 'msg': f'This Email ({email}) Has Been Used'}
@ -40,7 +41,7 @@ async def get_change_email_code(email: str, background_tasks: BackgroundTasks):
return {'code': 10400, 'msg': 'Send Verification Code Successfully'}
@router.post('/change/email')
@router.post('/change/email', response_model=BaseResponseModel)
async def change_email(changed_account: ChangedAccount):
is_correct = verify_code(changed_account.email, changed_account.code)
if not is_correct:
@ -51,7 +52,7 @@ async def change_email(changed_account: ChangedAccount):
return {'code': 10400, 'msg': 'Update Email Successfully'}
@router.get('/get/password_code')
@router.get('/get/password_code', response_model=BaseResponseModel)
async def get_change_password_code(email: str, background_tasks: BackgroundTasks):
if has_code(email):
return {'code': 10402, 'msg': f'Code of Email ({email}) Is Still Available'}
@ -60,7 +61,7 @@ async def get_change_password_code(email: str, background_tasks: BackgroundTasks
return {'code': 10400, 'msg': 'Send Verification Code Successfully'}
@router.post('/change/password')
@router.post('/change/password', response_model=BaseResponseModel)
async def change_password(changed_account: ChangedAccount):
is_correct = verify_code(changed_account.email, changed_account.code)
if not is_correct:

41
src/routers/user_profile.py
View File

@ -8,11 +8,15 @@ from fastapi.encoders import jsonable_encoder
from anyio import open_file
from ..crud import user_crud
from ..response_models.user_response import UserProfileResponse, UserAvatarResponse
from ..response_models.user_response import (
BaseResponseModel,
UserProfileResponse,
UserAvatarResponse,
)
from ..utils import static_file
router = APIRouter(prefix='/user_profile', tags=['user_profile'])
router = APIRouter(prefix="/user_profile", tags=["user_profile"])
class Uint8List(BaseModel):
@ -28,39 +32,43 @@ class ChangedProfile(BaseModel):
sign: Optional[str] = None
@router.get('/my', response_model=UserProfileResponse)
@router.get("/my", response_model=UserProfileResponse)
async def get_profile(id: str):
profile = await user_crud.select_profile(id)
return JSONResponse(
content=jsonable_encoder(
{'code': 10300, 'msg': 'Get My Profile Successfully', 'data': profile.to_dict_all()},
{
"code": 10300,
"msg": "Get My Profile Successfully",
"data": profile.to_dict(),
},
)
)
@router.get('/avatar')
@router.get("/avatar")
async def download_avatar(avatar_filename: str):
avatar_dir_path = static_file.create_avatar_dir()
avatar_dir_path = static_file.create_dir("avatars")
return FileResponse(avatar_dir_path / avatar_filename)
@router.post('/change/avatar', response_model=UserAvatarResponse)
@router.post("/change/avatar", response_model=UserAvatarResponse)
async def change_avatar(id: str, file: Uint8List):
avatar_dir_path = static_file.create_avatar_dir()
avatar_dir_path = static_file.create_dir("avatars")
avatar_filename = static_file.create_avatar_filename()
async with await open_file(avatar_dir_path / avatar_filename, 'wb') as f:
async with await open_file(avatar_dir_path / avatar_filename, "wb") as f:
await f.write(bytearray(file.file))
await user_crud.update_profile_avatar(id, avatar_filename)
return {'code': 10300, 'msg': 'Update Avatar Successfully', 'data': avatar_filename}
return {"code": 10300, "msg": "Update Avatar Successfully", "data": avatar_filename}
@router.post('/change/{aspect}')
@router.post("/change/{aspect}", response_model=BaseResponseModel)
async def change_profile(aspect: str, changed_profile: ChangedProfile):
match aspect:
case 'basic':
case "basic":
await user_crud.update_profile_basic(
changed_profile.id,
changed_profile.nickname,
@ -68,11 +76,10 @@ async def change_profile(aspect: str, changed_profile: ChangedProfile):
changed_profile.birthday,
changed_profile.gender,
)
case 'sign':
case "sign":
await user_crud.update_profile_sign(
changed_profile.id,
changed_profile.sign
changed_profile.id, changed_profile.sign
)
case _:
return {'code': 10301, 'msg': f'No /change/{aspect} Path'}
return {'code': 10300, 'msg': f'Update {aspect} Profile Successfully'}
return {"code": 10301, "msg": f"No /change/{aspect} Path"}
return {"code": 10300, "msg": f"Update {aspect} Profile Successfully"}

55
src/utils/static_file.py
View File

@ -1,15 +1,42 @@
import os
import random
from typing import Literal
from pathlib import Path
from datetime import datetime
from zipfile import ZipFile
alphabet = ['a', 'b', 'c', 'd', 'e', 'f', 'g', 'h', 'i', 'j', 'k', 'l', 'm',
'n', 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', 'w', 'x', 'y', 'z',
]
alphabet = [
"a",
"b",
"c",
"d",
"e",
"f",
"g",
"h",
"i",
"j",
"k",
"l",
"m",
"n",
"o",
"p",
"q",
"r",
"s",
"t",
"u",
"v",
"w",
"x",
"y",
"z",
]
def create_avatar_dir() -> Path:
avatar_dir_path = Path(os.getcwd()) / 'static' / 'avatars'
def create_dir(dir_name: str) -> Path:
avatar_dir_path = Path(os.getcwd()) / "static" / dir_name
if not avatar_dir_path.exists():
avatar_dir_path.mkdir()
return avatar_dir_path
@ -17,5 +44,19 @@ def create_avatar_dir() -> Path:
def create_avatar_filename() -> str:
timestamp = int(datetime.now().timestamp())
random_prefix = ''.join(random.choices(alphabet, k=8))
return f'{random_prefix}{timestamp}.png'
random_prefix = "".join(random.choices(alphabet, k=8))
return f"{random_prefix}{timestamp}.png"
def create_zip_file(filenames: list[str], file_type: Literal["avatars"]) -> Path:
zip_filename = f"temp_{file_type}_{round(datetime.now().timestamp())}.zip"
zip_dir = Path(os.getcwd()) / "static" / "temp"
file_dir = Path(os.getcwd()) / "static" / file_type
if not zip_dir.exists():
zip_dir.mkdir(parents=True)
with ZipFile(zip_dir / zip_filename, "w") as zip_file:
for filename in filenames:
zip_file.write(file_dir / filename, arcname=filename)
return zip_dir / zip_filename