geeker/student_manage_system
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/coffee_jiuye' into lon_students_merge_code_test

Browse Source
This commit is contained in:
Orangeeee
2026-09-22 10:48:43 +08:00
parent b4be2b083c 77de323412
commit 67c3cd0452
8 changed files with 445 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+2
View File
@@ -6,7 +6,9 @@ readme = "README.md"
requires-python = ">=3.13"
dependencies = [
"fastapi[standard]>=0.141.1",
"like>=1.7.1",
"pip>=26.2.1",
"pymysql>=1.2.3",
"sqlalchemy>=2.0.54",
"uvicorn>=0.53.0",
]
stu_jiuye/CURD_api.py
+113
View File
@@ -0,0 +1,113 @@
import datetime
from fastapi import APIRouter,Depends,HTTPException
from dao import *
from stu_jiuye.request import PostRequest, EmpResponse, AddRequest,AddResponse,CompRequest,CompResponse
from database import get_db
CURD=APIRouter()
@CURD.get('/emp',tags=['查询就业信息'])
def get_stuinfo( #查询学生就业信息
stuname:str| None = None
,stuclass:str| None = None
,compname:str|None=None
,min_salary:int=0
,max_salary:int|None=None
,somewhere:str|None=None
,employment_open_date:datetime.date|None=None
,offer_date:datetime.date|None=None
,db=Depends(get_db)
): #查询条件
r=get_emp_dao( #调用函数
stuname,
stuclass,
compname,
min_salary,
max_salary,
somewhere,
employment_open_date,
offer_date,
db
)
if r:
return r #成功返回
raise HTTPException(status_code=404, detail='名单不存在！') #失败返回错误信息
@CURD.post('/emp',response_model=EmpResponse,tags=['新增就业信息'])
def add_stuinfo( #新增学生就业信息
p:PostRequest,db=Depends(get_db)
): #调用请求体
d=p.model_dump() #解析成字典
r=add_employment_dao(o=d,db=db) #调用函数
if not r:
raise HTTPException(status_code=500, detail='服务器繁忙，请稍后添加！') #失败返回错误
return EmpResponse(totals=1, data=d) #成功返回响应体
@CURD.post('/emp/company',response_model=CompResponse,tags=['新增公司信息'])
def add_company( #新增公司表信息
c:CompRequest,db=Depends(get_db)
):
d = c.model_dump() #解析
r = add_company_dao(o=d, db=db) #调用添加公司函数
if not r:
raise HTTPException(status_code=500, detail='服务器繁忙，请稍后添加！') #失败返回错误信息
return CompResponse(totals=1, data=d) #成功返回响应体
@CURD.post('/emp/company/address',response_model=AddResponse,tags=['新增公司所在区域'])
def add_address( #新增公司地址区域信息
add:AddRequest,db=Depends(get_db)
):
d = add.model_dump() #解析
r = add_address_dao(o=d, db=db) #调用添加地址函数
if not r:
raise HTTPException(status_code=500, detail='服务器繁忙，请稍后添加！') #失败返回错误信息
return AddResponse(totals=1, data=d) #成功返回响应体
@CURD.put('/emp',tags=['更改就业信息'])
def updata_stuinfo( #更新数据模块
up:PostRequest
,id:int
,db=Depends(get_db)
): #根据id选择对象
d = up.model_dump(exclude_unset=True) #解析
r = update_emp_dao(id=id, update_data=d, db=db) #调用更新函数
if not r:
raise HTTPException(status_code=500, detail='没有更新！') #失败返回错误信息
return {'code': 200, 'totals': r, 'detail': '更新成功'} #成功返回信息
@CURD.delete('/emp/{id}',tags=['删除就业信息'])
def delete_stuinfo( #删除模块
id:int
,db=Depends(get_db)
): #根据id选择对象
rows = delete_emp_dao(id, db) #调用删除函数
if not rows:
raise HTTPException(status_code=500, detail='没有删除！') #失败返回错误信息
return {'code': 200, 'totals': rows, 'detail': '删除成功'} #成功返回信息
@CURD.get('/Top5',tags=['就业薪资Top5'])
def salarytop5(db=Depends(get_db)):
q = db.query(Employments,Company.compname) \
.join(Company, Employments.company_id == Company.id) \
.filter(Employments.is_deleted == 0)\
.order_by(Employments.salary.desc())\
.offset(0).limit(5) .all()
return [{'学生姓名':i.stuname,'学生班级':i.stuclass,'就业时间':i.offer_date,'公司':compname} for i,compname in q]
@CURD.get('/worktime',tags=['学生就业时常'])
def worktime(db=Depends(get_db)):
q = db.query(Employments).all()
return [{'学生姓名':i.stuname,'就业时常':f'{i.offer_date-i.employment_open_date}'} for i in q]
@CURD.get('/avgworktime',tags=['学生平均就业时常'])
def worktime(db=Depends(get_db)):
q = db.query(Employments)\
.filter(Employments.employment_open_date is not None)\
.all()
return [{'学生姓名':i.stuname,'平均就业时常':f'{(i.offer_date-i.employment_open_date)/len(q)}'} for i in q]
stu_jiuye/dao.py
+127
View File
@@ -0,0 +1,127 @@
from model import Employments,Address,Company
def add_employment_dao(o,db): #新增就业信息函数
try:
# r = db.query(Employments).filter(Employments.id == o['id'] ).all() #
o.pop('id', None) #摘除id字段，自增字段无需手动添加
o1 = Employments( **o) #实例化参数
db.add(o1) #添加数据进入表
except Exception as e: #失败返回
print(f'添加失败，{e}')
db.rollback() #回滚数据
return False
else:
db.commit() #提交事务，永久保存
return True
def add_company_dao(o,db): #添加公司信息函数
try:
o.pop('id', None) #摘除id字段，自增字段无需手动添加
o1 = Company( **o) #实例化参数
db.add(o1) #添加数据进入表
except Exception as e: #失败返回
print(f'添加失败，{e}')
db.rollback() #回滚数据
return False
else:
db.commit() #提交事务，永久保存
return True
def add_address_dao(o,db): #添加地址区域信息函数
try:
o.pop('id', None) #摘除id字段，自增字段无需手动添加
o1 = Address( **o) #实例化参数
db.add(o1) #添加数据进入表
except Exception as e: #失败返回
print(f'添加失败，{e}')
db.rollback() #回滚数据
return False
else:
db.commit() #提交事务，永久保存
return True
def update_emp_dao(id,update_data,db): #更新数据函数
try:
rows = db.query( Employments )\
.filter( Employments.id == id )\
.update( update_data ) #更新条件匹配id
except Exception as e:
print(f'更新失败，{e}') #返回失败信息
db.rollback() #数据回滚
return False
else:
db.commit() #提交事务，永久保存
return rows
def get_emp_dao( #获取学生就业信息函数
# stuid,
stuname
,stuclass
,compname
,min_salary
,max_salary
,somewhere
,employment_open_date
,offer_date
,db
# ,page
# ,page_size
): #根据选择条件获取信息
q = db.query(Employments,Company.compname,Address.somewhere)\
.join(Company,Employments.company_id==Company.id)\
.join(Address,Company.address_id==Address.id)\
.filter(Employments.is_deleted == 0) #连接公司表和地址表，筛选未被删除信息
# if stuid:
# q = q.filter( Employments.id == stuid )
if stuname: #根据学生姓名筛选
q = q.filter( Employments.stuname == stuname )
if stuclass:
q = q.filter(Employments.stuclass == stuclass)
if somewhere:
q = q.filter(Address.somewhere.like(f'%{somewhere}%'))
if compname: #根据公司姓名筛选
q=q.filter(Company.compname.like(f'%{compname}%'))
# r = q.offset( (page-1)*page_size ).limit( page_size ).all()
if min_salary is not None: #根据最小薪资筛选
q = q.filter(Employments.salary >= min_salary)
if max_salary is not None: ##根据最大薪资筛选
q = q.filter(Employments.salary <= max_salary)
if employment_open_date:
q = q.filter(Employments.employment_open_date == employment_open_date)
if offer_date:
q = q.filter(Employments.offer_date == offer_date)
r= q.all()
if r:
return [ {"id": i.id,
'学生姓名':i.stuname,
'学生班级':i.stuclass,
"公司": compname,
"公司区域":somewhere,
# "company_id": i.company_id,
"薪资":i.salary,
# "address_id": i.address_id,
"就业开放时间":i.employment_open_date,
"offer下发时间":i.offer_date,
"created_date": i.created_date,
"updated_date": i.updated_date,
"is_deleted":i.is_deleted } for i,compname,somewhere in r ] #返回信息列表
def delete_emp_dao(id,db): #删除函数
try:
rows = db.query(Employments)\
.filter(Employments.id == id)\
.update( {'is_deleted':1} ) #软删除，is_deleted返回1
except Exception as e: #失败返回信息
print(f'删除失败，{e}')
db.rollback() #数据回滚
rows = 0 #返回0
else:
db.commit() #提交事务，永久保存
return rows
stu_jiuye/database.py
+24
View File
@@ -0,0 +1,24 @@
from sqlalchemy import *
from sqlalchemy.orm import declarative_base,sessionmaker
db_url = "mysql+pymysql://root:123456@127.0.0.1:3306/ai0824_001?charset=utf8mb4"
engine = create_engine(db_url
, pool_size=50
, echo=True
)
Base = declarative_base()
Session = sessionmaker(bind=engine
,autoflush=False
,autocommit = False
)
def get_db():
db = Session()
try:
yield db
finally:
db.close()
stu_jiuye/main.py
+15
View File
@@ -0,0 +1,15 @@
from fastapi import FastAPI
from CURD_api import CURD
from database import Base,engine
import model
app=FastAPI()
Base.metadata.create_all(engine)
app.include_router(CURD,tags=['就业管理系统'])
if __name__=='__main__':
import uvicorn
uvicorn.run('main:app', host='127.0.0.1', port=12395)
stu_jiuye/model.py
+110
View File
@@ -0,0 +1,110 @@
from sqlalchemy import ForeignKey
from database import DATETIME,Base,Column,Integer,String,engine
from datetime import datetime
# class Student(Base):
# __tablename__ = 'Student'
# id = Column(Integer
# , primary_key=True
# , autoincrement=True
# # 外键
# ,comment='自增主键'
# )
# stuname = Column(String(50)
# ,nullable=False
# )
# stuid=Column(Integer
# ,nullable=False
# )
# stuclass=Column(String(50)
# ,nullable=False
# )
# created_date = Column(DATETIME
# ,default=datetime.now
# )
# updated_date = Column(DATETIME
# ,default=datetime.now
# ,onupdate=datetime.now
# )
class Employments(Base): #学生就业信息基类
__tablename__ = 'Employments' #表名
id = Column(Integer #主键，自增主键
, primary_key=True
, autoincrement=True
# , ForeignKey='student.id'
, comment='自增主键'
)
stuname = Column(String(50)) #学生姓名，冗余字段
stuclass = Column(String(50)) #学生班级，冗余字段
# compname=Column(String(50)
# ,nullable=False
# )
company_id = Column(Integer #公司id，外键匹配公司表
, ForeignKey('Company.id')
, nullable=False
)
# address_id = Column(Integer
# , ForeignKey('Address.id')
# , nullable=False
# )
salary = Column(Integer) #薪资
employment_open_date = Column(DATETIME) #就业开放时间
offer_date = Column(DATETIME) #offer下发时间
created_date = Column(DATETIME #创建数据时间
, default=datetime.now
)
updated_date = Column(DATETIME #更新数据时间
, default=datetime.now
, onupdate=datetime.now
)
is_deleted = Column(Integer, default=0) #是否为删除信息，默认为0
class Company(Base): #公司信息基类
__tablename__ = 'Company' #表名
id = Column(Integer
, primary_key=True
,autoincrement=True
) #主键id，自增主键
compname = Column(String(50) #公司姓名
, nullable=False #非空约束
)
address_id = Column(Integer #地址id,匹配地址表
, ForeignKey('Address.id') #外键地址表id
, nullable=False #非空约束
)
created_date = Column(DATETIME
, default=datetime.now
)
updated_date = Column(DATETIME
, default=datetime.now
, onupdate=datetime.now
)
is_deleted = Column(Integer, default=0)
class Address(Base):
__tablename__ = 'Address'
id = Column(Integer #地址id,自增主键
, primary_key=True
,autoincrement=True
)
somewhere=Column(String(50) #地址区域
, nullable=False
)
created_date = Column(DATETIME
, default=datetime.now
)
updated_date = Column(DATETIME
, default=datetime.now
, onupdate=datetime.now
)
is_deleted = Column(Integer, default=0)
stu_jiuye/request.py
+44
View File
@@ -0,0 +1,44 @@
from datetime import *
from pydantic import BaseModel, Field
class PostRequest(BaseModel): #学生就业信息请求体模型
# compname: str|None=None
stuname: str|None=None
stuclass: str|None=None
company_id:int|None=None
# address_id: int|None=None
salary: int|None=None
employment_open_date:date|None=None
offer_date:date|None=None
updated_date:date|None=None
is_deleted:int=0
class EmpResponse(BaseModel): #学生就业信息响应体
code: int = 200
detail: str = 'OK'
totals: int = 0
data: str | dict | tuple | list
class CompRequest(BaseModel): #公司信息请求体模型
compname:str
address_id: int | None = None
class CompResponse(BaseModel): #公司信息响应体
code: int = 200
detail: str = 'OK'
totals: int = 0
data: str | dict | tuple | list
class AddRequest(BaseModel): #地址请求体
somewhere:str
class AddResponse(BaseModel): #地址响应体
code: int = 200
detail: str = 'OK'
totals: int = 0
data: str | dict | tuple | list
uv.lock
Generated
+10
View File
@@ -396,6 +396,12 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" },
]
[[package]]
name = "like"
version = "1.7.1"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/0d/d1/5284ae5ae97876808622090d858098e11787469c601b679274e17bcb7b34/like-1.7.1.zip", hash = "sha256:2614e24923fb934955b6abeeda6d13db567d7901da44bc01480e14838618ded2", size = 925, upload-time = "2016-12-19T04:01:13.729Z" }
[[package]]
name = "markdown-it-py"
version = "4.2.0"
@@ -842,17 +848,21 @@ version = "0.1.0"
source = { virtual = "." }
dependencies = [
{ name = "fastapi", extra = ["standard"] },
{ name = "like" },
{ name = "pip" },
{ name = "pymysql" },
{ name = "sqlalchemy" },
{ name = "uvicorn" },
]
[package.metadata]
requires-dist = [
{ name = "fastapi", extras = ["standard"], specifier = ">=0.141.1" },
{ name = "like", specifier = ">=1.7.1" },
{ name = "pip", specifier = ">=26.2.1" },
{ name = "pymysql", specifier = ">=1.2.3" },
{ name = "sqlalchemy", specifier = ">=2.0.54" },
{ name = "uvicorn", specifier = ">=0.53.0" },
]
[[package]]