api/e_api.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter,HTTPException,Depends
+from database import Session
+from model.e_model import Employment
+from schema.e_schema import EmploymentRequest
+from dao.e_dao import get_employment_dao,wq_employment_dao,upd_employment_dao,del_employment_dao
+employment_api = APIRouter()
+
+
+def get_db():
+    db = Session()
+    try:
+        yield db
+    finally:
+        db.close()
+@employment_api.get('/',summary='查询就业信息')
+def get_employment(db=Depends(get_db)):
+    r = get_employment_dao(db)
+    return {'code':200,'detail':'查询成功!','data':r}
+
+
+@employment_api.post('/',summary='新增就业信息')
+def add_employment(e:EmploymentRequest,db=Depends(get_db)):
+        d= e.model_dump()
+        r = wq_employment_dao(d,db)
+        return {'code':200,'detail':'添加成功!','total':r}
+#
+#
+@employment_api.put('/',summary='更新就业信息')
+def update_employment(e:EmploymentRequest,db=Depends(get_db)):
+    e1 = e.model_dump()
+    r = upd_employment_dao(e1,db)
+    return {'code':200,'msg':'更新成功','total':r}
+#
+@employment_api.delete('/',summary='删除就业信息')
+def delete_employment(stu_id:str,db=Depends(get_db)):
+    r =del_employment_dao(stu_id,db)
+    return {'code':200,'totals':r,'detail':'删除成功!'}

dao/e_dao.py

@@ -0,0 +1,52 @@
+from http.client import HTTPException
+
+from model.e_model import Employment
+from fastapi import HTTPException
+
+def get_employment_dao(db):
+    try:
+        r = db.query(Employment).all()
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500, detail=f'查询失败:str(e)')
+    else:
+        db.commit()
+    return [{'stu_id':i.stu_id,'class_name':i.class_name} for i in r ]
+
+
+def wq_employment_dao(o,db):
+    try:
+        o1 = Employment(**o)
+        db.add(o1)
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500, detail=f'添加失败:str(e)')
+    else:
+        db.commit()
+        return 1
+
+def upd_employment_dao(e,db):
+    try:
+        rows = db.query(Employment).filter(Employment.stu_id == e['stu_id']).update(e)
+    except:
+        db.rollback()
+        raise HTTPException(status_code=500, detail='更新异常，请稍后再执行！')
+    else:
+        db.commit()
+        return 1
+
+def del_employment_dao(stu_id,db):
+    try:
+        row = db.query(Employment).filter(Employment.stu_id == stu_id).all()
+        if not row:
+            raise HTTPException(status_code=404,detail='记录不存在')
+        for i in row:
+            if i.deleted == 0:
+                db.query(Employment).filter(Employment.stu_id == i.stu_id).update({'deleted':1})
+    except Exception as e:
+        db.rollback()
+        raise HTTPException(status_code=500,detail=f'删除失败:{str(e)}')
+    else:
+        db.commit()
+        return 1
+

model/e_model.py

@@ -0,0 +1,45 @@
+from sqlalchemy import *
+from datetime import datetime
+from sqlalchemy.orm import declarative_base,sessionmaker
+
+db_url = "mysql+pymysql://root:123456@127.0.0.1:3306/ai0824_stu?charset=utf8"
+
+engine = create_engine(db_url, pool_size=100,echo=False)
+
+Base = declarative_base()
+
+class Employment(Base):
+    __tablename__ = 's_employment'
+    id = Column(Integer
+                , primary_key=True
+                , autoincrement=True
+                , comment='编号,自增主键')
+    stu_id = Column(String(20)
+                  # , ForeingKey('s_student.stu_id')
+                  , unique=True
+                  , nullable=True
+                  , comment='学生学号，唯一')
+    class_name = Column(String(32))
+    company = Column(String(100))
+    salary = Column(Integer
+                    , default=0)
+    open_date = Column(DATETIME
+                     , default=datetime.now())
+    offer_date = Column(DATETIME
+                     , default=datetime.now())
+    work_date = Column(DATETIME
+                     , default=datetime.now()
+                     , onupdate=datetime.now)
+    is_delete = Column(Integer
+                   , default=0
+                     )
+
+
+Session = sessionmaker( bind = engine, autoflush = False, autocommit = False )
+
+def get_db():
+    db = Session()
+    try:
+        yield db
+    finally:
+        db.close()

schema/e_schema.py

@@ -0,0 +1,13 @@
+from pydantic import BaseModel
+from typing import Optional
+
+class EmploymentRequest(BaseModel):
+    stu_id: str|None = None
+    class_name: str|None = None
+    company: Optional[str] = None
+    salary: Optional[int] = None
+
+
+
+
+