python-fastapi-mm-zhcs-yj-api/common/db/db_user.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
from sqlalchemy.orm import Session
from sqlalchemy import text, exists, and_, or_, not_
from sqlalchemy.sql import func
from database import get_db
from models import *
from extensions import logger
from utils import *
from common.enc import mpfun

def get_user_name_by_id(db: Session, user_id: int) -> str:
    row = db.query(SysUser).filter(SysUser.user_id == user_id).first()
    if row is not None:
        return row.user_name
    
    return ""

def get_user_id_by_phonenumber(db: Session, phonenumber: str) -> int:
    # 手机号码已加密
    row = db.query(SysUser).filter(and_(SysUser.phonenumber == mpfun.enc_data(phonenumber), SysUser.del_flag == '0')).first()
    if row is not None:
        return row.user_id

    return -1

def get_user_id_by_yzyaccount(db: Session, yzyaccount: str) -> int:
    row = db.query(SysUser).filter(and_(SysUser.yzy_account == yzyaccount, SysUser.del_flag == '0')).first()
    if row is not None:
        return row.user_id

    return -1

def get_nick_name_by_id(db: Session, user_id: int) -> str:
    row = db.query(SysUser).filter(SysUser.user_id == user_id).first()
    if row is not None:
        return row.nick_name
    
    return ""

def get_user_info(db: Session, user_id: int) -> SysUser:
    row = db.query(SysUser).filter(SysUser.user_id == user_id).first()
    return row