add emal interface

README.MD

@@ -15,3 +15,11 @@ https://fastapi.tiangolo.com/zh/tutorial/bigger-applications/
 3. 后端采用redis，mysql
 
 
+#### route分类
+> common
+    mail  -clearn
+> deploy
+    aliyun cdn
+
+> ops
+> data_collect

config/config.py

@@ -0,0 +1,15 @@
+from pydantic import BaseConfig
+from typing import Optional
+import os
+
+
+class Settings(BaseConfig):
+    redis_host: str = "192.168.100.30"  # reids 服务器IP
+    redis_port: int = 6379  # redis 端口
+    redis_db: int = 2   # redis db
+    mail_server: str = "smtp.exmail.qq.com"  # 邮箱server
+    mail_user: str = "ops@kingsome.cn"  # 邮箱用户名
+    mail_pswd: Optional[str] = os.getenv('mail_pswd')  # 邮箱密码
+
+
+settings = Settings()

internal/admin.py

@@ -1,7 +0,0 @@
-from fastapi import APIRouter
-router = APIRouter()
-
-
-@router.post("/")
-async def update_admin():
-    return {"message": "Admin getting schwifty"}

main.py

@@ -1,17 +1,21 @@
-# cd .. && uvicorn myops.main:app --reload
+# uvicorn main:app --host=127.0.0.1 --port=8000 --reload
 from fastapi import Depends, FastAPI, BackgroundTasks
-from .dependencies import get_query_token, get_token_header
-from .routers import items, users
-from .internal import admin
+from dependencies import get_query_token, get_token_header
+# from routers import items, users
+# from internal import admin
+from ops.common import common
 from typing import Optional
+from scripts.common.redis import get_redis_pool
+# import sys
+# sys.path.append('.')
 
 tags_metadata = [
+    # {
+    #     "name": "common",
+    #     "description": "Operations with users. The **login** logic is also here.",
+    # },
     {
-        "name": "users",
-        "description": "Operations with users. The **login** logic is also here.",
-    },
-    {
-        "name": "items",
+        "name": "common",
         "description": "Manage items. So _fancy_ they have their own docs.",
         "externalDocs": {
             "description": "Items external docs",
@@ -20,34 +24,34 @@ tags_metadata = [
     },
 ]
 
-# app = FastAPI(dependencies=[Depends(get_token_header)], title="My ops project",
-#               description="This is some interface for ops in kingsome!", version="1.0.1")
-
 
 app = FastAPI(dependencies=[Depends(get_token_header)],
               openapi_tags=tags_metadata)
 
 
-app.include_router(users.router)
-app.include_router(items.router)
-app.include_router(
-    admin.router,
-    prefix="/admin",
-    tags=["admin"],
-    dependencies=[Depends(get_token_header)],
-    responses={418: {"description": "I`m a teapot"}}
-)
-
-app.get("/")
+@app.on_event("startup")
+async def startup_event():
+    app.state.redis = await get_redis_pool()
 
 
+@app.on_event("shutdown")
+async def shutdown_event():
+    app.state.redis.close()
+    await app.state.redis.wait_close()
+
+
+app.include_router(common.router)
+
+
+@app.get("/")
 async def root():
     return {"message": "Hello Bigger Applications!"}
 
 
-def write_log(message: str):
+def write_log(message: str) -> True:
     with open("log.txt", mode="a") as log:
         log.write(message)
+    return True
 
 
 def get_query(background_tasks: BackgroundTasks, q: Optional[str] = None):
@@ -62,3 +66,9 @@ async def send_notification(email: str, background_tasks: BackgroundTasks, q: st
     message = f"message to {email} \n"
     background_tasks.add_task(write_log, message)
     return {"message": "Message sent"}
+
+
+if __name__ == '__main__':
+    import uvicorn
+    uvicorn.run(app='main:app', host="127.0.0.1",
+                port=8010, reload=True, debug=True)

ops/common/common.py

@@ -0,0 +1,41 @@
+# pip install fastapi-mail
+from fastapi import APIRouter, BackgroundTasks, UploadFile, File, Form
+from fastapi_mail import FastMail, MessageSchema, ConnectionConfig
+from typing import List
+from pydantic import BaseModel, EmailStr
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from config.config import settings
+router = APIRouter()
+
+
+class EmailSchema(BaseModel):
+    email: List[EmailStr]
+    body: str
+    subject: str
+
+
+MAIL_CONF = ConnectionConfig(
+    MAIL_USERNAME=settings.mail_user,
+    MAIL_PASSWORD=settings.mail_pswd,
+    MAIL_FROM=settings.mail_user,
+    MAIL_PORT='465',
+    MAIL_SERVER=settings.mail_server,
+    MAIL_TLS=False,
+    MAIL_SSL=True,
+    # USE_CREDENTIALS=True
+)
+
+
+@router.post("/common/email")
+async def simple_send(email: EmailSchema) -> JSONResponse:
+    message = MessageSchema(
+        subject=email.dict().get("subject"),
+        # List of recipients, as many as you can pass
+        recipients=email.dict().get("email"),
+        body="<html>{}</html>".format(email.dict().get("body")),
+        subtype="html"
+    )
+    fm = FastMail(MAIL_CONF)
+    await fm.send_message(message)
+    return JSONResponse(status_code=200, content={"message": "email has been sent"})

ops/wjtx/wjtx.py

@@ -0,0 +1,30 @@
+from fastapi import APIRouter, Depends, HTTPException
+from ...dependencies import get_token_header
+
+router = APIRouter(
+    prefix="/wjtx",
+    tags=["wjtx"],
+    dependencies=[Depends(get_token_header)],
+    responses={404: {"description": "Not found"}},
+
+)
+
+
+@router.get("/")
+async def read_items():
+    return {"msg": "1"}
+
+
+# @router.get("/{itemid}")
+# async def read_item(item_id: str):
+#     if item_id not in fake_item_db:
+#         raise HTTPException(status_code=404, detail="Item not found")
+#     return {"name": fake_item_db[item_id]["name"], "item_id": item_id}
+
+
+# @router.put("/{item_id}", tags=["custom"], responses={403: {"description": "Operation forbidden"}})
+# async def update_item(item_id: str):
+#     if item_id != "pulmbus":
+#         raise HTTPException(
+#             status_code=403, detail="You can only update the item plumbus")
+#     return {"item_id": item_id, "name": "The great plumbus"}

routers/items.py

@@ -1,32 +0,0 @@
-from fastapi import APIRouter, Depends, HTTPException
-from ..dependencies import get_token_header
-
-router = APIRouter(
-    prefix="/items",
-    tags=["item"],
-    dependencies=[Depends(get_token_header)],
-    responses={404: {"description": "Not found"}},
-
-)
-
-fake_item_db = {"plumbus": {"name": "Plumbus", "gun": {"name": "Portal Gun"}}}
-
-
-@router.get("/")
-async def read_items():
-    return fake_item_db
-
-
-@router.get("/{itemid}")
-async def read_item(item_id: str):
-    if item_id not in fake_item_db:
-        raise HTTPException(status_code=404, detail="Item not found")
-    return {"name": fake_item_db[item_id]["name"], "item_id": item_id}
-
-
-@router.put("/{item_id}", tags=["custom"], responses={403: {"description": "Operation forbidden"}})
-async def update_item(item_id: str):
-    if item_id != "pulmbus":
-        raise HTTPException(
-            status_code=403, detail="You can only update the item plumbus")
-    return {"item_id": item_id, "name": "The great plumbus"}

routers/users.py

@@ -1,17 +0,0 @@
-from fastapi import APIRouter
-router = APIRouter()
-
-
-@router.get('/users/', tags=["users"])
-async def read_users():
-    return [{"username": "Rick"}, {"username": "Morty"}]
-
-
-@router.get("/user/me", tags=["users"])
-async def read_user_me():
-    return {"username": "fakecurrentuser"}
-
-
-@router.get("/user/{username}", tags=["users"])
-async def read_user(username: str):
-    return {"username": username}

scripts/common/redis.py

@@ -0,0 +1,14 @@
+from aioredis import create_redis_pool, Redis
+from config.config import settings
+
+# settings = {
+#     "redis_host": "192.168.100.30",
+#     "redis_port": 6379,
+#     "redis_db": 1,
+
+# }
+
+
+async def get_redis_pool() -> Redis:
+    redis = await create_redis_pool("redis://{ip}:{port}/{db}?encoding=utf-8".format(ip=settings.redis_host, port=settings.redis_port, db=settings.redis_db))
+    return redis

test.py

@@ -0,0 +1,30 @@
+from fastapi.testclient import TestClient
+from ops.common.common import EmailSchema
+from main import app
+import time
+import pdb
+import json
+client = TestClient(app)
+
+
+def test_read_main():
+    response = client.get("/")
+    assert response.status_code == 200
+    assert response.json() == {"message": "Hello Bigger Applications!"}
+
+
+def test_post_mail():
+    data = {"email": ['pengtao@kingsome.cn'], "body": 'test message!',
+            "subject": 'ttt {}'.format(time.time())}
+    response = client.post("/common/email", headers={"X-Token": "fake-super-scret-token", "accept": "application/json", "Content-Type": "application/json"},
+                           data=json.dumps(data),
+                           )
+    # data=EmailSchema(**data).json()
+    print(response)
+    # pdb.set_trace()
+    assert response.status_code == 200
+    assert response.json() == {"message": "email has been sent"}
+
+
+if __name__ == "__main__":
+    test_post_mail()