fast api init

fastapi-demo/config/__init__.py

@@ -10,13 +10,9 @@ import os
 from concurrent.futures import ThreadPoolExecutor
 from configparser import ConfigParser
 from logging.handlers import TimedRotatingFileHandler
+from os.path import dirname
 
-import aredis
-import redis
-import requests
 import uvloop
-from aiokafka import AIOKafkaProducer
-from elasticsearch import Elasticsearch
 from requests import adapters
 
 adapters.DEFAULT_POOLSIZE = 100000
@@ -31,71 +27,11 @@ class CfgBaseInit(object):
     start_mode: 项目启动模式
     """
     config = ConfigParser()
-
     config.read(os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.ini"))
-    project_name = config.get("CONFIG", "project_name")
-    project_path = os.path.join(os.path.abspath(os.path.dirname(__file__)).split(project_name)[0], project_name)
-    start_model = config.get("CONFIG", "start_model").upper()
-    __authentication = config.get("CONFIG", "authentication")
-    authentication = json.loads(__authentication if "'" not in __authentication else __authentication.replace("'", '"'))
-    __headers = config.get("CONFIG", "headers")
-    headers = json.loads(__headers if "'" not in __headers else __headers.replace("'", '"'))
-    bind = config.get(section=f"PROJECT-{start_model}", option="host")
-    _executor = ThreadPoolExecutor(max_workers=50000)  # 线程池
-    _async_kafka_session = None
-
-    kafka_server = config.get(f"BEHAVE-{start_model}", "kafka_server")
-
-    @staticmethod
-    def get_my_event_loop():
-        """
-            获取当前 loop 循环事件管理对象
-        :return:
-        """
-        event_loop = asyncio.get_event_loop()
-        # asyncio.set_event_loop(loop=event_loop)
-        return event_loop
-
-    @classmethod
-    def executor_submit(cls, *args, **kwargs):
-        """
-            改写 executor.submit 方法，加入默认的异常捕获功能；
-            使用方式与 原来的 executor.submit 相同
-        :param args:
-        :param kwargs:
-        :return:
-        """
-        if asyncio.iscoroutinefunction(args[0]):
-            loop = asyncio.new_event_loop()
-            if not loop.is_running():
-                executor_future = loop.run_until_complete(args[0](args[1:]))  # 处理 future 函数 or 对象
-                return executor_future
-        else:
-            thread_name_prefix = kwargs.get("thread_name_prefix", "")
-            if not thread_name_prefix:
-                thread_name_prefix = args[0].__name__
-            if "thread_name_prefix" in kwargs:
-                del kwargs['thread_name_prefix']
-                cls._executor._thread_name_prefix = thread_name_prefix
-            executor_future = cls._executor.submit(*args, **kwargs)  # 处理 普通 function；
-            # 线程任务增加任务名称前缀，与线程名称相同.
-            # 默认为 function.__name__
-            executor_future.__setattr__("name", thread_name_prefix)
-            executor_future.add_done_callback(cls._executor_callback)
-            return executor_future
 
-    @classmethod
-    def _executor_callback(cls, worker):
-        """
-            任务池中任务异常捕获回调函数
-            抛出异常
-        :param worker: 任务
-        :return:
-        """
-        worker_exception = worker.exception()
-        if worker_exception:
-            logger.exception("Worker return exception: {}".format(worker_exception))
-            raise worker_exception
+    project_name = config.get("CONFIG", "project_name")
+    project_path = dirname(dirname(__file__))
+    bind = config.get("CONFIG", "host")
 
 
 class Logger(object):
@@ -115,124 +51,3 @@ class Logger(object):
 
 
 logger = Logger.logger
-
-
-class RedisInit(CfgBaseInit):
-    """
-        Redis 对象初始化
-    """
-    section = "REDIS-{}".format(CfgBaseInit.start_model.upper())
-    host = CfgBaseInit.config.get(section=section, option="host")
-    port = CfgBaseInit.config.get(section=section, option="port")
-    db = CfgBaseInit.config.get(section=section, option="db")
-
-    host_bak = CfgBaseInit.config.get(section=section, option="host_bak")
-    port_bak = CfgBaseInit.config.get(section=section, option="port_bak")
-    db_bak = CfgBaseInit.config.get(section=section, option="db_bak")
-    Pool = redis.ConnectionPool(host=host, port=port, db=db, max_connections=None, decode_responses=True,
-                                socket_keepalive=False)
-    Pool_bak = redis.ConnectionPool(host=host_bak, port=port_bak, db=db_bak, max_connections=None,
-                                    decode_responses=True)
-    conn = redis.Redis(connection_pool=Pool)
-
-
-class AsyncRedisInit(CfgBaseInit):
-    """
-        redis 异步连接池初始化
-    pipeline_demo：
-        async def pipeline_test():
-            conn = aredis.StrictRedis(connection_pool=AsyncRedisInit.Pool)
-            p = await conn.pipeline()
-            await p.hgetall("process_cache\0015f371f42f286b2eb461e75a437162c6e@cccxx@abrs_reco@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@0@0_0")
-            a = await p.execute()
-            return a
-
-    client_demo:
-        async def conn_test():
-            conn = aredis.StrictRedis(connection_pool=AsyncRedisInit.Pool)
-            a = await conn.hgetall("process_cache\0015f371f42f286b2eb461e75a437162c6e@cccxx@abrs_reco@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@d751713988987e9331980363e24189ce@0@0_0")
-            return a
-    """
-    section = "REDIS-{}".format(CfgBaseInit.start_model.upper())
-    host = CfgBaseInit.config.get(section=section, option="host")
-    port = CfgBaseInit.config.get(section=section, option="port")
-    try:
-        password = CfgBaseInit.config.get(section=section, option="password")
-    except Exception:
-        password = None
-    db = CfgBaseInit.config.get(section=section, option="db")
-
-    host_bak = CfgBaseInit.config.get(section=section, option="host_bak")
-    port_bak = CfgBaseInit.config.get(section=section, option="port_bak")
-    try:
-        password_bak = password
-    except Exception:
-        password_bak = None
-    db_bak = CfgBaseInit.config.get(section=section, option="db_bak")
-    # todo 不能使用链接池，链接池不能自动释放 conn
-    Pool = aredis.ConnectionPool(host=host, port=port, db=db,
-                                 password=password, max_connections=None, decode_responses=True,
-                                 socket_keepalive=False)
-    Pool_bak = aredis.ConnectionPool(host=host_bak, port=port_bak, db=db_bak, password=password_bak,
-                                     max_connections=None, decode_responses=True,
-                                     socket_keepalive=False)
-    conn = aredis.StrictRedis(connection_pool=Pool)
-
-
-class ElasticSearchInit(CfgBaseInit):
-    """
-        ElasticSearch 对象初始化
-    """
-    section = "ES-{}".format(CfgBaseInit.start_model.upper())
-    index = CfgBaseInit.config.get(section=section, option="index")
-    fc_index = CfgBaseInit.config.get(section=section, option="fc_index")
-    try:
-        es_timeout = float(CfgBaseInit.config.get(section=section, option="timeout"))
-    except Exception as e:
-        es_timeout = 10
-    es_client = Elasticsearch(hosts=[CfgBaseInit.config.get(section=section, option="host")], timeout=es_timeout)
-    fc_es_client = Elasticsearch(hosts=CfgBaseInit.config.get(section=section, option="fc_host").split(","),
-                                 timeout=es_timeout)
-
-
-class BehaviorAPI(CfgBaseInit):
-    """
-        行为数据对接接口 Behavior Api 初始化
-    """
-    section = "BEHAVIOR_API-{}".format(CfgBaseInit.start_model.upper())
-    url = CfgBaseInit.config.get(section=section, option="url")
-
-    @staticmethod
-    def push_behavior_log(data):
-        data = json.dumps(data)
-        r = requests.post(BehaviorAPI.url, data=data, headers=CfgBaseInit.headers)
-        return r
-
-
-class AsyncKafka(CfgBaseInit):
-    """异步请求模块
-    """
-    _async_kafka_session = None
-
-    @classmethod
-    async def __set_async_session(cls):
-        producer = AIOKafkaProducer(loop=cls.get_my_event_loop(),
-                                    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
-                                    bootstrap_servers=CfgBaseInit.kafka_server)
-        await producer.start()
-        cls._async_kafka_session = producer
-        return producer
-
-    @classmethod
-    async def async_push(cls, payload):
-        """异步请求接口
-        """
-        request_session = cls._async_kafka_session or await cls.__set_async_session()
-        print(request_session)
-        await request_session.send_and_wait("eip_rec_behave", payload)
-
-
-if __name__ == '__main__':
-    a = BehaviorAPI.push_behavior_log({"a": 1})
-    print(a)
-    print(a.content)

fastapi-demo/config/config.ini

@@ -1,162 +1,11 @@
 [CONFIG]
-project_name = eip_feed_behave
-;project_name = eip_ai_asgi_py387
+project_name = test
 headers = {"content-Type": "application/json"}
-authentication = {
-                 "hezhiguo": "hezhiguo",
-                 "eip-ai": "eip-ai"
-                 }
-; dev: 开发(组内自测)：redis-6379
-; test: 测试环境, ES index 不同
-; pro：预发、生产
-start_model = dev
+host = 0.0.0.0:8089
 
 [LOG]
 ; 绝对路径
-log_path = ../logs/eip_feed_behave/info_rec.log
+log_path = info.log
 ; h、d、m、s，时、天、分、秒
 when = d
 backupCount = 30
-
-[PROJECT-DEV]
-host = 0.0.0.0:8089
-
-[PROJECT-TEST]
-host = 0.0.0.0:8088
-
-[PROJECT-PRO]
-host = 0.0.0.0:80
-
-[ES-DEV]
-host = 10.10.3.83:9200
-fc_host = 10.10.3.83:9200,10.10.100.198:9200
-user = eip
-pwd = eip-pass
-index = zdfx_item_test
-;fc_index = hippo_text_doc_27_515
-fc_index = hippo_text_doc_321_0615
-timeout = 60
-
-[ES-TEST]
-host = 10.10.3.83:9200
-fc_host = 10.10.3.83:9200,10.10.100.198:9200
-user = eip
-pwd = eip-pass
-index = zdfx_item_test
-;fc_index = hippo_text_doc_27_515
-fc_index = hippo_text_doc_321_0615
-timeout = 60
-
-[ES-PRO]
-host = 10.10.3.83:9200
-fc_host = 10.10.3.83:9200,10.10.100.198:9200
-user = eip
-pwd = eip-pass
-index = eip_inference_feed_item
-;fc_index = hippo_text_doc_27_515
-fc_index = hippo_text_doc_321_0615
-timeout = 60
-
-[MYSQL-PRO]
-url = mysql+mysqlconnector://root:qwertyuiop@10.10.100.228:8137/eip_ai?auth_plugin=mysql_native_password
-hit_log_table = eip_ai_hit_log
-
-[REDIS-DEV]
-host = 127.0.0.1
-port = 6379
-db = 0
-expire = 1000
-host_bak = 127.0.0.1
-port_bak = 6379
-db_bak = 15
-expire_bak = 1000
-pool_size = 10
-users = {
-        "hezhiguo": "hezhiguo",
-        "eip-ai": "eip-ai"
-        }
-
-[REDIS-TEST]
-host = 127.0.0.1
-port = 6379
-db = 0
-expire = 1000
-host_bak = 127.0.0.1
-port_bak = 6379
-db_bak = 15
-expire_bak = 1000
-pool_size = 10
-users = {
-        "hezhiguo": "hezhiguo",
-        "eip-ai": "eip-ai"
-        }
-
-[REDIS-PRO]
-host = 127.0.0.1
-port = 6379
-db = 0
-expire = 1000
-host_bak = 127.0.0.1
-port_bak = 6379
-db_bak = 15
-expire_bak = 1000
-pool_size = 10
-users = {
-        "hezhiguo": "hezhiguo",
-        "eip-ai": "eip-ai"
-        }
-
-[KAFKA_BEHAVIOR-DEV]
-user_name = eip-ai
-dept_name = eip-ai
-host = 1.kafka.adh:9092,2.kafka.adh:9092,3.kafka.adh:9092,4.kafka.adh:9092,5.kafka.adh:9092,6.kafka.adh:9092,7.kafka.adh:9092
-topic = log_server_stream
-is_cluster = 0
-reset_offset_on_start = 0
-fetch_wait_max_ms = 30000
-queued_max_messages = 2000
-
-[KAFKA_BEHAVIOR-TEST]
-user_name = eip-ai
-dept_name = eip-ai
-host = 1.kafka.adh:9092,2.kafka.adh:9092,3.kafka.adh:9092,4.kafka.adh:9092,5.kafka.adh:9092,6.kafka.adh:9092,7.kafka.adh:9092
-topic = log_server_stream
-is_cluster = 0
-reset_offset_on_start = 0
-fetch_wait_max_ms = 30000
-queued_max_messages = 2000
-
-[KAFKA_BEHAVIOR-PRO]
-user_name = eip-ai
-dept_name = eip-ai
-host = 1.kafka.adh:9092,2.kafka.adh:9092,3.kafka.adh:9092,4.kafka.adh:9092,5.kafka.adh:9092,6.kafka.adh:9092,7.kafka.adh:9092
-topic = log_server_stream
-is_cluster = 0
-reset_offset_on_start = 0
-fetch_wait_max_ms = 30000
-queued_max_messages = 2000
-
-[BEHAVIOR_API-DEV]
-url = http://115.159.79.118:20001/log/mlamp/eip/dw/EIP/ABRS/behavior_test
-;url = http://192.168.64.20:5000/abrs_update
-[BEHAVIOR_API-TEST]
-url = http://115.159.79.118:20001/log/mlamp/eip/dw/EIP/ABRS/behavior_test
-;url = http://192.168.64.20:5000/abrs_update
-
-[BEHAVIOR_API-PRO]
-url = http://115.159.79.118:20001/log/mlamp/eip/dw/EIP/ABRS/behavior
-
-[BLOOM_FILTER]
-;盐值；
-SALTS = ("1", "2", "3", "4")
-;length; 长度必须小于 alts 长度，且在 3个以上
-SALTS_LEN_LESS_1 = 3
-;hash list; 2**9 * 2**20 * 2**3  # 512MB*1024*1024*8　：MB->bit位
-hash_list = 4294967296
-hash_func_name = md5
-
-[BEHAVE-DEV]
-kafka_server = eip-kafka-2.qa.mlamp.cn
-
-[BEHAVE-PRO]
-kafka_server = eip-ckafka-3.mlamp.cn

fastapi-demo/controller/base_controller.py

@@ -5,6 +5,6 @@ from fastapi.responses import JSONResponse
 router = APIRouter(prefix="/base", tags=['base'])
 
 
-@router.get("/check", name="服务状态校验", tags=['BASE'])
+@router.get("/check", name="服务状态校验")
 async def listener():
     return JSONResponse({"message": "success", "statusCode": 200}, status_code=200)

fastapi-demo/controller/user_controller.py

@@ -0,0 +1,10 @@
+# -*- coding:utf-8 -*-
+from fastapi import APIRouter
+from fastapi.responses import JSONResponse
+
+router = APIRouter(prefix="/user", tags=['user'])
+
+
+@router.get("/name", name="name")
+async def listener():
+    return JSONResponse({"message": "hi", "statusCode": 200}, status_code=200)

fastapi-demo/debug.py

@@ -2,4 +2,4 @@
 import uvicorn
 
 if __name__ == '__main__':
-    uvicorn.run(app='api_main:app', host="0.0.0.0", port=8087, reload=True, debug=True)
+    uvicorn.run(app='main:app', host="0.0.0.0", port=8087, reload=True, debug=True)

fastapi-demo/main.py

@@ -3,6 +3,7 @@ import json
 import os
 import time
 
+import fastapi
 from fastapi import FastAPI
 from fastapi import HTTPException, Request
 from fastapi.exceptions import RequestValidationError
@@ -11,25 +12,20 @@ from fastapi.responses import PlainTextResponse
 from pydantic import BaseModel  # 参数校验
 from starlette.exceptions import HTTPException as StarletteHTTPException
 from starlette.middleware.sessions import SessionMiddleware
-from starlette.responses import JSONResponse
+from starlette.responses import JSONResponse, RedirectResponse
 from starlette.staticfiles import StaticFiles
 
 from config import logger, CfgBaseInit
 
-app = FastAPI(title="EIP REC ENGINE ", description="通用系统 ", version="v 0.0.0")
+app = FastAPI(title="project name ", description="通用系统 ", version="v 0.0.0")
 
 app.routes.pop(1)  # drop default docs_url router
 
 # 添加 session 中间键，使项目中可以使用session
-app.add_middleware(SessionMiddleware, secret_key='123456')
-
-# 跨域白名单, 解决跨域问题
-origins = [
-    "*",
-]
+app.add_middleware(SessionMiddleware, secret_key='123456hhh')
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=origins,
+    allow_origins=["*", ],
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
@@ -64,21 +60,12 @@ async def validation_exception_handler(request, exc):
 @app.middleware("http")
 async def add_process_time_header(request: Request, call_next):
     """接口响应中间键; 当前只支持 http 请求"""
-    logger.info(f'【 {request.scope.get("path").replace("/", ".").lstrip(".")} 】API query start...')
+    logger.info(f'【 {request.scope.get("path")} 】API query start...')
     start_time = time.time()
     response = await call_next(request)
     process_time = time.time() - start_time
     response.headers["API-Process-Time"] = f"{process_time} seconds"
-    response.headers["service"] = "EIP REC ENGINE"
-    response.headers["Request-Client"] = str(request.scope.get("client")[0]) + ":" + str(
-        request.scope.get("client")[1])
-    response.headers["User-Agent"] = str(request.headers.get("user-agent"))
-    logger.info(f'Request-Client is :【 {response.headers.get("Request-Client")} 】')
-    logger.info(f'Request-Client User-Agent is :【 {response.headers.get("User-Agent")} 】')
-    request.scope.get("path").replace("/", "")
-    logger.info(f'【 {request.scope.get("path")} 】API '
-                f'Process-Time:【 {process_time} seconds 】...')
-
+    logger.info(f'【 {request.scope.get("path")} 】API, Time:【 {process_time} seconds 】...')
     return response
 
 
@@ -99,7 +86,7 @@ for file in os.listdir(os.path.join(CfgBaseInit.project_path, "controller")):
 
 @app.get("/")
 def index():
-    return {"message": "你已经正确创建api"}
+    return RedirectResponse(url="/static/index.html")
 
 
 # http://localhost:8000/query?uid=1
@@ -123,7 +110,7 @@ class People(BaseModel):
     salary: float
 
 
-# jsonbody请求
+# json body请求
 @app.post("/insert")
 def insert(people: People):
     msg = f"名字：{people.name},年龄{people.age}"

fastapi-demo/static/index.html

@@ -0,0 +1,10 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <title>Title</title>
+</head>
+<body>
+<h1>hello</h1>
+</body>
+</html>

tmp/PyWebIO/demo01.py

@@ -0,0 +1,21 @@
+from pywebio import start_server
+from pywebio.input import input, FLOAT
+from pywebio.output import put_text
+
+def bmi():
+    height = input("请输入你的身高(cm)：", type=FLOAT)
+    weight = input("请输入你的体重(kg)：", type=FLOAT)
+
+    BMI = weight / (height / 100) ** 2
+
+    top_status = [(14.9, '极瘦'), (18.4, '偏瘦'),
+                  (22.9, '正常'), (27.5, '过重'),
+                  (40.0, '肥胖'), (float('inf'), '非常肥胖')]
+
+    for top, status in top_status:
+        if BMI <= top:
+            put_text('你的 BMI 值: %.1f，身体状态：%s' % (BMI, status))
+            break
+
+if __name__ == '__main__':
+    start_server(bmi, port=8080)

tmp/tmp.py

@@ -1,3 +1,5 @@
+a = 123
+i = 1234
 import aiohttp
 import asyncio