feat: docker本地调试:

- 1. import不同系统语法不一致
- 2. mysql允许远程访问

cloudcontainers/app/crud.py

@@ -0,0 +1,56 @@
+from sqlalchemy.orm import Session
+from sqlalchemy import select
+
+
+from .  import models, schemas
+
+
+# 删除
+def delete_user(db: Session, user_id: int):
+    db.query(models.User).where(models.User.id == user_id).delete()
+    db.commit()
+    # db.refresh(db_user)
+
+
+# 更新
+def update_user(db: Session, user_id: int):
+    db_test=db.query(models.User).filter(models.User.id == user_id).update({"email":'111'},synchronize_session=False)
+    db.commit()
+    # db_test=db.query(models.User).filter(models.User.id == user_id).first()
+    # db_test.email='12345@qq.com'
+    # db.commit()
+
+
+# 查询
+def get_user(db: Session, user_id: int):
+    return db.query(models.User).filter(models.User.id == user_id).first()
+
+
+def get_user_by_email(db: Session, email: str):
+    return db.query(models.User).filter(models.User.email == email).first()
+
+
+def get_users(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.User).offset(skip).limit(limit).all()
+
+# 新增
+def create_user(db: Session, user: schemas.UserCreate):
+
+    fake_hashed_password = user.password + "notreallyhashed"
+    db_user = models.User(email=user.email, hashed_password=fake_hashed_password,is_active=False)
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user
+
+
+def get_items(db: Session, skip: int = 0, limit: int = 100):
+    return db.query(models.Item).offset(skip).limit(limit).all()
+
+
+def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
+    db_item = models.Item(**item.dict(), owner_id=user_id)
+    db.add(db_item)
+    db.commit()
+    db.refresh(db_item)
+    return db_item

cloudcontainers/app/database.py

@@ -0,0 +1,15 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import pymysql
+
+# SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
+# 从docker访问ip需要 mysql服务端允许外部访问=> 参考地址:https://blog.csdn.net/weixin_42599091/article/details/125224850
+SQLALCHEMY_DATABASE_URL = "mysql+pymysql://root:Xiao4669805@192.168.31.245:3306/test"
+
+# SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False} #sqlite
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

cloudcontainers/app/main.py

@@ -0,0 +1,114 @@
+from fastapi import FastAPI,Header,Depends,HTTPException
+import requests
+import json
+from enum import Enum
+from typing import Union
+from pydantic import BaseModel
+
+
+from sqlalchemy.orm import Session
+from . import crud, models, schemas
+from .database import SessionLocal, engine
+models.Base.metadata.create_all(bind=engine)
+
+
+app = FastAPI()
+
+# Dependency
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+        
+class User(BaseModel):
+    user_id: str
+
+class ModelName(str, Enum):
+    alexnet = "alexnet"
+    resnet = "resnet"
+    lenet = "lenet"
+
+class Item(BaseModel):
+    name: str
+    description: Union[str, None] = None
+    price: float
+    tax: Union[float, None] = None
+
+@app.post("/delete_user")
+async def delete(user:User,db: Session = Depends(get_db)):
+    db_user = crud.delete_user(db,user_id=user.user_id)
+    # print('db_user'+db_user)
+    return {"message": "Hello delete_user"}
+
+
+@app.post("/update_user")
+async def update(user:User,db: Session = Depends(get_db)):
+    db_user = crud.update_user(db,user_id=user.user_id)
+    return {"message": "Hellupdate_usero World"}
+
+
+
+@app.get("/")
+async def root():
+    return {"message": "Hello World"}
+
+@app.post("/users/", response_model=schemas.User)
+def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
+    db_user = crud.get_user_by_email(db, email=user.email)
+    if db_user:
+        raise HTTPException(status_code=400, detail="Email already registered")
+    return crud.create_user(db=db, user=user)
+
+# 飞书通知
+@app.get("/feishu")
+async def root():
+    headers = {'Content-Type': 'application/json'}
+    # https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd
+    try:
+        r = requests.post('https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd',
+                        data=json.dumps({'msg_type': 'text', "content": {"text": "cityNew有新用户注册"}}), headers=headers)
+    except(e):
+        print('e')
+    return {"message": "触发飞书webhook"}
+
+# 飞书通知
+@app.post("/nihao")
+async def root():
+    headers = {'Content-Type': 'application/json'}
+    # https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd
+    try:
+        r = requests.post('https://open.feishu.cn/open-apis/bot/v2/hook/5fb1b44c-3446-42ef-9b3a-dd37731567dd',
+                        data=json.dumps({'msg_type': 'text', "content": {"text": "cityNew有新用户注册"}}), headers=headers)
+    except(e):
+        print('e')
+    return {"message": "nihao"}
+
+# post请求
+@app.post("/testPost")
+async def root():
+    return {"message": "test-post"}
+
+# 路径参数
+@app.get("/path/{pathName}")
+async def root(pathName:ModelName):
+    return {"message": pathName}
+
+# 查询参数
+@app.get("/items/")
+async def read_item(skip: int = 0,  limit: Union[int, None] = None):
+    fake_items_db = [{"item_name": "Foo"}, {"item_name": "Bar"}, {"item_name": "Baz"}]
+    # return fake_items_db[skip : skip + limit]
+    return f'{skip}+{limit}'
+
+# 请求体
+@app.post("/postitems/")
+async def create_item(item: Item):
+    return item
+
+
+# header参数
+@app.get("/items/header")
+async def read_items(user_agent: Union[str, None] = Header(default=None)):
+    return {"User-Agent": user_agent}

cloudcontainers/app/models.py

@@ -0,0 +1,26 @@
+from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
+from sqlalchemy.orm import relationship
+
+from .database import Base
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True, index=True)
+    email = Column(String(1000), unique=True, index=True)
+    hashed_password = Column(String(1000))
+    is_active = Column(Boolean, default=True)
+
+    items = relationship("Item", back_populates="owner")
+
+
+class Item(Base):
+    __tablename__ = "items"
+
+    id = Column(Integer, primary_key=True, index=True)
+    title = Column(String(1000), index=True)
+    description = Column(String(1000), index=True)
+    owner_id = Column(Integer, ForeignKey("users.id"))
+
+    owner = relationship("User", back_populates="items")

cloudcontainers/app/schemas.py

@@ -0,0 +1,35 @@
+from pydantic import BaseModel
+
+
+class ItemBase(BaseModel):
+    title: str
+    description: str | None = None
+
+
+class ItemCreate(ItemBase):
+    pass
+
+
+class Item(ItemBase):
+    id: int
+    owner_id: int
+
+    class Config:
+        orm_mode = True
+
+
+class UserBase(BaseModel):
+    email: str
+
+
+class UserCreate(UserBase):
+    password: str
+
+
+class User(UserBase):
+    id: int
+    is_active: bool
+    items: list[Item] = []
+
+    class Config:
+        orm_mode = True