fastAPi搭建框架实现登录 -------day04
apis一登录模型建表modelsclass JobSeeker(models.Model): # 主键ID自增 id fields.IntField(pkTrue) # 手机号唯一 mobile fields.CharField(max_length20, uniqueTrue) # 密码 password fields.CharField(max_length128) # 头像 avatar fields.CharField(max_length500, nullTrue) # 昵称 nickname fields.CharField(max_length100, nullTrue) # 邮箱 email fields.CharField(max_length100, nullTrue) # 微信号 wechat fields.CharField(max_length100, nullTrue) # 创建时间 created_at fields.DatetimeField(auto_now_addTrue) # 更新时间 updated_at fields.DatetimeField(auto_nowTrue) class Meta: table job_seeker table_description 求职者信息表账号密码实现登录操作servicesclass JobSeekerService: staticmethod async def login_mobile_password(login_info:JobSeekerLoginRequest): jobseeker await JobSeeker.get_or_none(mobilelogin_info.mobile) # 判断手机号 if jobseeker is None: raise Exception(手机号不存在) # 验证密码 if jobseeker.password ! login_info.password: raise Exception(密码错误) # 生成Token access_token, refresh_token create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f生成 access_token{access_token}) logger.info(f生成 refresh_token{refresh_token}) return { jobseeker: jobseeker, access_token: access_token, refresh_token: refresh_token }apisjob_seeker.post(/job_seeker, summary登录,description登录) async def login_by_mobile_password(login_request:JobSeekerLoginRequest): res await JobSeekerService.login_mobile_password(login_request) return { code: 200, message: 登录成功, data: res }手机号验证码实现操作servicesstaticmethod async def send_sms_code(mobile: str): 1判断手机号是否存在 2生成验证码 3发送验证码 4保存验证码 :return: jobseeker await JobSeeker.get_or_none(mobilemobile) if jobseeker is None: raise Exception(手机号不存在) code get_random_code() logger.info(f生成验证码{code}) key fboss-api:login:sms:{mobile} redis_client.set(key, code, ex60*2) return code staticmethod async def register_job_seeker(login_request:JobSeekerSmsCodeRequest): 1判断手机号是否存在 2,redis中是否存在验证码并且一样登录成功生成token删除redis中的验证码 3不一样登录失败 :return: jobseeker await JobSeeker.get_or_none(mobilelogin_request.mobile) if jobseeker is None: raise Exception(手机号不存在) key fboss-api:login:sms:{login_request.mobile} redis_code redis_client.get(key) if redis_code is None: raise Exception(验证码已过期) if redis_code ! login_request.code: raise Exception(验证码错误) access_token, refresh_token create_tokens(str(jobseeker.id), jobseeker.mobile) logger.info(f生成 access_token{access_token}) logger.info(f生成 refresh_token{refresh_token}) # 删除 redis_client.delete(key) return { jobseeker: jobseeker, access_token: access_token, refresh_token: refresh_token }apisjob_seeker.post(/send_sms_code/{mobile}, summary发送验证码,description发送验证码) async def send_sms_code(mobile:str): res await JobSeekerService.send_sms_code(mobile) return { code: 200, message: 发送成功, data: res } job_seeker.post(/register_job_seeker, summary手机号验证码登录,description手机号验证码登录) async def register_job_seeker(login_request:JobSeekerSmsCodeRequest): res await JobSeekerService.register_job_seeker(login_request) return { code: 200, message: 登录成功, data: res }随机6位数验证码获取工具coreimport random def get_random_code(length: int 6): code .join(random.choices(123456789, klength)) return code接收后端传的个人用户信息servicesstaticmethod async def get_info(authorization: str): 1验证token 2获取用户信息 :return: job_seeker_id get_info(authorization) job_seeker await JobSeeker.get_or_none(idjob_seeker_id) job_seeker.nickname f用户:{job_seeker.mobile} if job_seeker.nickname is None else job_seeker.nickname return job_seekerapis 前端传递token,放在headers authorizationBearer token job_seeker.get(/me, summary获取当前用户信息,description获取当前用户信息) async def get_me(job_seeker_idDepends(get_info)): logger.info(f通过token获取当前用户信息ID:{job_seeker_id}) job_seeker await JobSeekerService.get_info_by_id(job_seeker_id) return { code: 200, message: 获取成功, data: job_seeker }文件上传阿里云下载依赖 pip install oss在utils里配置三、文件上传阿里云 1、配置 依赖下载 pip install oss utils/oss_util # oss_tool.py import oss2 import os from uuid import uuid4 from typing import List, Optional, Tuple from datetime import timedelta from oss2.exceptions import OssError from app.core.logging import logger from app.config.settings import settings class AliyunOSSTool: pip install oss22.19.1 pip install python-multipart 阿里云OSS工具类基于oss2封装 封装OSS的文件上传、下载、删除、签名URL生成等核心操作 def __init__( self, access_key_id: Optional[str] None, access_key_secret: Optional[str] None, endpoint: Optional[str] None, bucket_name: Optional[str] None ): 初始化OSS工具类 :param access_key_id: 阿里云AccessKey ID :param access_key_secret: 阿里云AccessKey Secret :param endpoint: OSS地域节点如oss-cn-beijing.aliyuncs.com :param bucket_name: OSS Bucket名称 说明若参数为None则从环境变量中读取对应配置 # 1. 加载配置优先传入参数其次环境变量 self.access_key_id access_key_id or settings.ALIYUN_OSS_ACCESS_KEY_ID self.access_key_secret access_key_secret or settings.ALIYUN_OSS_ACCESS_KEY_SECRET self.endpoint endpoint or settings.ALIYUN_OSS_ENDPOINT self.bucket_name bucket_name or settings.ALIYUN_OSS_BUCKET_NAME # 2. 校验配置完整性 self._check_config() # 3. 初始化OSS Auth和Bucket对象 self.auth oss2.Auth(self.access_key_id, self.access_key_secret) self.bucket oss2.Bucket( self.auth, self.endpoint, self.bucket_name, connect_timeout30, ) def _check_config(self) - None: 校验配置是否完整缺失则抛出异常 logger.info(OSS配置校验开始) logger.info(fAccessKey ID: {self.access_key_id}) logger.info(fAccessKey Secret: {* * min(len(self.access_key_secret or ), 8)}...) logger.info(fEndpoint: {self.endpoint}) logger.info(fBucket Name: {self.bucket_name}) # 检查配置是否为空或占位符 def is_valid_config(value: Optional[str]) - bool: 检查配置值是否有效非空且不是占位符 if not value: return False # 检查是否是占位符 placeholder_values [KEY, SECRET, your-, your_] return not any(placeholder in str(value) for placeholder in placeholder_values) required_configs { AccessKey ID: self.access_key_id, AccessKey Secret: self.access_key_secret, Endpoint: self.endpoint, Bucket Name: self.bucket_name } missing_configs [name for name, value in required_configs.items() if not is_valid_config(value)] if missing_configs: raise ValueError(fOSS配置缺失或无效{, .join(missing_configs)}请检查.env.dev文件或环境变量) def _generate_unique_filename(self, filename: str, oss_path: str) - str: 生成唯一的OSS文件路径避免文件名重复覆盖 :param filename: 原始文件名 :param oss_path: OSS存储目录 :return: 唯一的oss_file_key如uploads/abc123-test.jpg # 提取文件后缀 file_suffix os.path.splitext(filename)[1] # 生成UUID前缀保证唯一性 uuid_prefix str(uuid4())[:8] # 构造唯一文件名 unique_filename f{uuid_prefix}-{os.path.splitext(filename)[0]}{file_suffix} # 构造完整OSS文件路径 return f{oss_path.rstrip(/)}/{unique_filename} def upload_single_file(self, file_content: bytes, filename: str, oss_path: str uploads/, unique_filename: bool True) - Tuple[bool, dict]: 上传单个文件到OSS :param file_content: 文件字节内容 :param filename: 原始文件名 :param oss_path: OSS存储目录默认uploads/ :param unique_filename: 是否生成唯一文件名默认True避免覆盖 :return: (上传是否成功, 结果数据字典) try: # 1. 构造OSS文件路径 if unique_filename: oss_file_key self._generate_unique_filename(filename, oss_path) else: oss_file_key f{oss_path.rstrip(/)}/{filename} # 2. 调用oss2上传文件 result self.bucket.put_object(oss_file_key, file_content) # 3. 构造返回结果 if result.status 200: file_access_url fhttps://{self.bucket_name}.{self.endpoint}/{oss_file_key} return True, { filename: filename, oss_file_key: oss_file_key, access_url: file_access_url, file_size: len(file_content), etag: result.etag } else: return False, {error: fOSS返回非200状态码{result.status}} except OssError as e: return False, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, {error: f文件上传失败{str(e)}} def upload_multiple_files(self, files_data: List[Tuple[bytes, str]], oss_path: str uploads/, unique_filename: bool True) - List[dict]: 批量上传文件到OSS :param files_data: 文件数据列表每个元素为(文件字节内容, 原始文件名) :param oss_path: OSS存储目录默认uploads/ :param unique_filename: 是否生成唯一文件名默认True :return: 批量上传结果列表 upload_results [] for file_content, filename in files_data: success, result_data self.upload_single_file(file_content, filename, oss_path, unique_filename) upload_results.append({ filename: filename, status: success if success else failed, data: result_data }) return upload_results def download_file(self, oss_file_key: str) - Tuple[bool, Optional[oss2.models.GetObjectResult], Optional[dict]]: 从OSS下载文件返回文件流对象 :param oss_file_key: OSS文件完整路径如uploads/abc123-test.jpg :return: (下载是否成功, 文件流对象, 错误信息字典) try: # 1. 检查文件是否存在 if not self.bucket.object_exists(oss_file_key): return False, None, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 读取文件流oss2流式返回避免大文件内存溢出 file_object self.bucket.get_object(oss_file_key) return True, file_object, None except OssError as e: return False, None, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, None, {error: f文件下载失败{str(e)}} def delete_file(self, oss_file_key: str) - Tuple[bool, dict]: 从OSS删除指定文件 :param oss_file_key: OSS文件完整路径 :return: (删除是否成功, 结果数据字典) try: # 1. 检查文件是否存在 if not self.bucket.object_exists(oss_file_key): return False, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 调用oss2删除文件 result self.bucket.delete_object(oss_file_key) # 3. 验证删除结果 if result.status 204: return True, {oss_file_key: oss_file_key, message: 文件删除成功} else: return False, {error: fOSS返回非204删除状态码{result.status}} except OssError as e: return False, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, {error: f文件删除失败{str(e)}} def generate_sign_url(self, oss_file_key: str, expire_seconds: int 3600, method: str GET) - Tuple[ bool, Optional[str], dict]: 生成OSS文件临时签名URL私有Bucket必备用于临时访问/下载 :param oss_file_key: OSS文件完整路径 :param expire_seconds: URL有效期默认3600秒/1小时 :param method: 访问方法GET下载/访问PUT上传 :return: (生成是否成功, 签名URL, 结果数据字典) try: # 1. 检查文件是否存在可选根据业务需求决定是否开启 if not self.bucket.object_exists(oss_file_key): return False, None, {error: f文件 {oss_file_key} 不存在于OSS中} # 2. 调用oss2生成签名URL sign_url oss2.sign_url( self.auth, self.endpoint, self.bucket_name, oss_file_key, expire_seconds, methodmethod ) return True, sign_url, {expire_seconds: expire_seconds, method: method} except OssError as e: return False, None, {error: fOSS操作异常{str(e)}, status_code: e.status} except Exception as e: return False, None, {error: f签名URL生成失败{str(e)}}
)
)
)
