问题：使用fastapi框架时，如何进行文件上传？

方法：

方法一

import shutil
from fastapi import FastAPI, File, UploadFile  

app = FastAPI()

@app.post("/uploadfile/")
async def create_upload_file(file: UploadFile = File(...)):
    # 这里可以添加保存文件的逻辑
    # 例如，使用 Python 的内置 open() 函数将文件保存到服务器
    # 注意：出于安全考虑，您应该验证文件类型和大小
    with open(file.filename, "wb") as buffer:  
        shutil.copyfileobj(file.file, buffer)  

    return {"filename": file.filename}

阅读全文

问题：python如何判断字符串是否包含汉字？

方法：

def is_Chinese(word):
    for ch in word:
        if '\u4e00' <= ch <= '\u9fff':
            return True
    return False
print(is_Chinese("测试"))

阅读全文

问题：因为使用整数作为url查询参数，所以容易被遍历查询

解决：对整数键进行加密

方法：

# 加密
('加密%s' % id).encode('utf-8').hex()

# 简单的加解密
def decryt_hex(h):
    s = bytes.fromhex(h).decode('utf-8')
    return s[2:]

阅读全文

问题：通过接口进行hive查询，太慢，使用fastapi框架，本想使用BackgroundTasks，但hive没有异步处理框架。查询到说是可以使用asyncio方法实现，但试了下后，发现不可以

解决：使用rq队列，通过队列方式处理hive查询，然后进行回调

注意：

1、jobs中的任务文件，引入config时注意路径，问题，可能需要把config所在目录加入项目环境变量

2、rq执行有时间限制，配置redis时需要配置下

# 增加超时时间为300秒  
q = Queue(connection=redis_conn, default_timeout=300)
或者
job = q.enqueue(my_function, args=(arg1, arg2), timeout=300)

问题：rqworker 运行时报错ERROR:root:Authentication required. 使用的是redis作为存储，因为redis设置了密码

解决：可以通过--url参数设置redis连接链接；也可以通过export设置redis连接链接为环境变量

方法：

1、rq连接redis

from rq import Queue
from redis import StrictRedis
from .config import config

redis_config = { 
    'host': config['RQ_SVR'],
    'db': config['RQ_DB'],
    'password': config['RQ_PASS']
    }   
redis_url = 'redis://:{password}@{host}:6379/{db}'.format(**redis_config)
redis_conn = StrictRedis.from_url(redis_url)
q = Queue(connection=redis_conn)

阅读全文