# PyLiteFlux
PyLiteFlux 是一个轻量级的Python Web/TCP服务器框架，完全基于Python标准库实现，不依赖任何第三方包。它支持HTTP和TCP协议，提供了简单而强大的API来构建网络应用。适合用于学习和理解Python后端框架的概念和实现原理。
## 特性
- 支持HTTP和TCP服务器
- 支持TCP长连接和消息推送
- 支持后台任务处理
- 内置中间件系统
- 简单的路由注册
- 支持配置管理
- 支持请求上下文
## 安装
```bash
pip install pyliteflux
```
## 快速开始
### 基本使用
```python
from pyliteflux import Server, Route
server = Server()
@Route.register("/hello", method="GET")
def hello(request):
return {"message": "Hello, World!"}
server.run()
```
### TCP长连接和消息推送示例
下面是一个简单的聊天系统示例，展示了如何使用TCP长连接和消息推送功能：
```python
from pyliteflux import Server, Route
import time
server = Server()
server.config.tcp_keep_alive = True # 启用TCP长连接
# HTTP接口：发送消息给指定客户端
@Route.register("/hello", method="GET")
def hello(request):
# 从查询参数中获取目标client_id
target_client = request.query_params.get("client_id")
def heavy_task():
time.sleep(5) # 模拟耗时操作
server.push_message("Hello via HTTP GET!", target_client)
request.bg.add(heavy_task)
return {
"message": "Hello via HTTP GET!",
"target_client": target_client,
"query_params": request.query_params
}
# TCP接口：列出所有活跃连接
@Route.register("clients", protocol="tcp")
def list_clients(request):
clients = list(server._active_connections.keys())
return {
"message": "Active TCP clients",
"clients": clients,
"count": len(clients)
}
# TCP接口：订阅消息
@Route.register("subscribe", protocol="tcp")
def subscribe_tcp(request):
client_id = f"{request.client_address[0]}:{request.client_address[1]}"
return {"message": "subscribed", "client_id": client_id}
server.run()
```
## 主要功能
### HTTP服务器
- 支持GET、POST、PUT、DELETE方法
- 支持路由参数
- 支持查询参数
- 支持JSON和表单数据
### TCP服务器
- 支持长连接
- 支持消息推送
- 支持JSON和XML数据格式
- 支持自定义命令
### 后台任务
- 支持异步任务处理
- 任务队列管理
- 线程池执行
### 中间件系统
- 请求处理中间件
- 响应处理中间件
- 上下文管理中间件
### 中间件和全局上下文示例
#### 中间件使用
中间件可以处理请求和响应，实现认证、日志、性能监控等功能：
```python
from pyliteflux import Server, Route, Middleware
# 定义认证中间件
class AuthMiddleware(Middleware):
def process_request(self, request):
# 检查认证token
token = request.headers.get("Authorization")
if not token:
return {"error": "Unauthorized"}
# 将认证信息存储在g对象中
g.user = self.validate_token(token)
return None # 继续处理请求
def process_response(self, request, response):
# 处理响应
if isinstance(response, dict):
response["authenticated"] = hasattr(g, "user")
return response
def validate_token(self, token):
# 实际项目中这里应该验证token
return {"id": 1, "name": "user1"}
# 定义日志中间件
class LogMiddleware(Middleware):
def process_request(self, request):
# 记录请求开始时间
g.start_time = time.time()
return None
def process_response(self, request, response):
# 计算请求处理时间
duration = time.time() - g.start_time
Logger.info(f"Request processed in {duration:.2f}s")
return response
# 使用中间件
server = Server()
server.middleware_manager.add_middleware(AuthMiddleware())
server.middleware_manager.add_middleware(LogMiddleware())
# 在路由中使用g对象
@Route.register("/user", method="GET")
def get_user(request):
if hasattr(g, "user"):
return {
"message": "User info",
"user": g.user,
"request_id": g.request_id # 每个请求的唯一ID
}
return {"error": "User not found"}
server.run()
```
#### 全局上下文使用
g对象是请求级别的全局对象，可以在整个请求过程中共享数据：
```python
from pyliteflux import Server, Route, g
server = Server()
# 在中间件中设置数据
class DataMiddleware(Middleware):
def process_request(self, request):
g.db = Database() # 假设的数据库连接
g.cache = Cache() # 假设的缓存连接
return None
def process_response(self, request, response):
# 清理资源
if hasattr(g, "db"):
g.db.close()
if hasattr(g, "cache"):
g.cache.close()
return response
# 在路由中使用g对象
@Route.register("/data", method="GET")
def get_data(request):
# 使用g对象中的资源
data = g.db.query("SELECT * FROM users")
cached_data = g.cache.get("users")
return {
"data": data,
"cached": cached_data,
"request_id": g.request_id
}
# 在后台任务中使用g对象
@Route.register("/async", method="GET")
def async_task(request):
def background_job():
# 注意：后台任务有自己的g对象实例
g.task_id = uuid.uuid4()
Logger.info(f"Background task {g.task_id} started")
request.bg.add(background_job)
return {"message": "Task scheduled"}
server.run()
```
#### 实际应用场景
1. 用户认证和授权：
```python
class AuthMiddleware(Middleware):
def process_request(self, request):
token = request.headers.get("Authorization")
if token:
g.user = self.authenticate(token)
g.permissions = self.get_permissions(g.user)
return None
@Route.register("/admin", method="GET")
def admin_panel(request):
if not hasattr(g, "permissions") or "admin" not in g.permissions:
return {"error": "Access denied"}, 403
return {"message": "Welcome, admin!"}
```
2. 性能监控：
```python
class PerformanceMiddleware(Middleware):
def process_request(self, request):
g.start_time = time.time()
g.sql_queries = []
return None
def process_response(self, request, response):
duration = time.time() - g.start_time
metrics = {
"duration": duration,
"sql_queries": len(g.sql_queries),
"memory_usage": self.get_memory_usage()
}
Logger.info(f"Performance metrics: {metrics}")
return response
```
3. 请求跟踪：
```python
class TracingMiddleware(Middleware):
def process_request(self, request):
g.trace_id = str(uuid.uuid4())
g.span_id = str(uuid.uuid4())
g.traces = []
return None
def process_response(self, request, response):
if isinstance(response, dict):
response["trace_id"] = g.trace_id
response["traces"] = g.traces
return response
@Route.register("/trace", method="GET")
def traced_endpoint(request):
g.traces.append({"event": "processing", "time": time.time()})
result = process_data()
g.traces.append({"event": "completed", "time": time.time()})
return {"data": result}
```
这些示例展示了中间件和g对象的强大功能：
1. 请求前后的处理
2. 资源的自动管理
3. 请求级别的数据共享
4. 性能监控和日志记录
5. 认证和授权
6. 请求跟踪和调试
中间件和g对象的组合使用可以大大提高开发效率和代码质量。
## 配置选项
```python
server = Server()
server.config.update(
http_host="127.0.0.1",
http_port=8000,
tcp_host="127.0.0.1",
tcp_port=8001,
tcp_keep_alive=True, # 启用TCP长连接
info=True # 启用日志
)
```
## API文档
### 路由装饰器
```python
@Route.register(path, method="GET", protocol="http")
```
- path: 路由路径
- method: HTTP方法（GET/POST/PUT/DELETE）
- protocol: 协议类型（http/tcp）
### 服务器方法
```python
server.run() # 启动服务器
server.stop() # 停止服务器
server.push_message(message, client_id=None) # 推送消息
```
### 请求对象
```python
request.query_params # 查询参数
request.form_data # 表单数据
request.json_data # JSON数据
request.headers # 请求头
request.bg.add(task) # 添加后台任务
```
### 响应格式
PyLiteFlux支持多种响应格式，包括普通JSON响应、流式响应和自定义JSON响应：
#### 1. 普通JSON响应
````python
@Route.register("/hello", method="GET")
def hello(request):
return {"message": "Hello, World!"} # 自动转换为JSON响应
````
#### 2. 流式响应
流式响应适用于大数据传输或需要实时推送数据的场景：
````python
from pyliteflux import StreamResponse
@Route.register("/stream", method="GET")
def stream_data(request):
def generate_data():
for i in range(10):
time.sleep(1) # 模拟数据生成
yield f"data: {i}\n\n"
return StreamResponse(
generate_data(),
content_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive"
}
)
# 带参数的流式响应
@Route.register("/download", method="GET")
def download_file(request):
def file_reader():
with open("large_file.txt", "rb") as f:
while chunk := f.read(8192):
yield chunk
return StreamResponse(
file_reader(),
content_type="application/octet-stream",
headers={
"Content-Disposition": "attachment; filename=large_file.txt"
}
)
````
#### 3. HTTPJSONResponse响应
当需要更细粒度地控制JSON响应时，可以使用HTTPJSONResponse：
````python
from pyliteflux import HTTPJSONResponse
@Route.register("/api/user", method="GET")
def get_user(request):
user_data = {"id": 1, "name": "John"}
return HTTPJSONResponse(
data=user_data,
status=200,
headers={
"X-Custom-Header": "value",
"Access-Control-Allow-Origin": "*"
}
)
# 错误响应示例
@Route.register("/api/error", method="GET")
def error_response(request):
return HTTPJSONResponse(
data={"error": "Not Found", "code": "404"},
status=404,
headers={"X-Error-Type": "NotFound"}
)
# 自定义序列化示例
@Route.register("/api/custom", method="GET")
def custom_serialization(request):
class CustomEncoder(json.JSONEncoder):
def default(self, obj):
if isinstance(obj, datetime):
return obj.isoformat()
return super().default(obj)
data = {
"timestamp": datetime.now(),
"message": "Custom serialization"
}
return HTTPJSONResponse(
data=data,
json_encoder=CustomEncoder,
headers={"Content-Type": "application/json"}
)
````
#### 4. 组合使用示例
````python
@Route.register("/api/complex", method="GET")
def complex_response(request):
response_type = request.query_params.get("type", "json")
if response_type == "stream":
def stream():
for i in range(5):
yield json.dumps({"count": i}) + "\n"
return StreamResponse(stream(), content_type="application/x-ndjson")
elif response_type == "error":
return HTTPJSONResponse(
{"error": "Bad Request"},
status=400,
headers={"X-Error-Details": "Invalid type"}
)
else:
return {"message": "Regular JSON response"} # 自动转换为JSON
````
## 参数使用说明
### 服务器配置参数
服务器实例支持多种配置参数，可以通过config对象进行设置：
```python
from pyliteflux import Server
server = Server()
# 方式1：通过属性设置
server.config.http_host = "127.0.0.1"
server.config.http_port = 8000
server.config.tcp_keep_alive = True
# 方式2：通过update方法批量设置
server.config.update(
http_host="127.0.0.1",
http_port=8000,
tcp_host="127.0.0.1",
tcp_port=8001,
tcp_keep_alive=True,
info=True
)
```
配置参数说明：
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| http_host | str | "127.0.0.1" | HTTP服务器主机地址 |
| http_port | int | 8000 | HTTP服务器端口 |
| tcp_host | str | "127.0.0.1" | TCP服务器主机地址 |
| tcp_port | int | 8001 | TCP服务器端口 |
| tcp_keep_alive | bool | False | 是否启用TCP长连接 |
| info | bool | True | 是否启用日志输出 |
### 路由参数
路由装饰器支持多种参数配置：
```python
from pyliteflux import Route
# HTTP路由
@Route.register("/hello", method="GET") # 基本GET请求
@Route.register("/user", method="POST") # POST请求
@Route.register("/user/{id}", method="GET") # 带路径参数的路由
@Route.register("/files/*", method="GET") # 通配符路由
# TCP路由
@Route.register("command", protocol="tcp") # TCP命令路由
```
路由参数说明：
| 参数 | 类型 | 默认值 | 说明 |
|------|------|--------|------|
| path | str | 必填 | 路由路径 |
| method | str | "GET" | HTTP方法（GET/POST/PUT/DELETE） |
| protocol | str | "http" | 协议类型（http/tcp） |
### 请求参数
HTTP请求可以通过多种方式传递参数：
```python
# 1. 查询参数
@Route.register("/search", method="GET")
def search(request):
keyword = request.query_params.get("keyword")
page = request.query_params.get("page", "1")
return {"keyword": keyword, "page": page}
# 2. 路径参数
@Route.register("/user/{user_id}/posts/{post_id}", method="GET")
def get_post(request):
user_id = request.path_params["user_id"]
post_id = request.path_params["post_id"]
return {"user_id": user_id, "post_id": post_id}
# 3. 表单数据
@Route.register("/upload", method="POST")
def upload(request):
file_name = request.form_data.get("file_name")
content = request.form_data.get("content")
return {"file_name": file_name}
# 4. JSON数据
@Route.register("/api/data", method="POST")
def handle_json(request):
data = request.json_data
return {"received": data}
```
### TCP连接参数
TCP连接相关的参数和使用方式：
```python
# 1. 启用TCP长连接
server = Server()
server.config.tcp_keep_alive = True
# 2. 发送消息给特定客户端
@Route.register("/push", method="GET")
def push_message(request):
client_id = request.query_params.get("client_id")
message = request.query_params.get("message", "Hello!")
server.push_message(message, client_id)
return {"status": "sent"}
# 3. 广播消息给所有客户端
@Route.register("/broadcast", method="GET")
def broadcast(request):
message = request.query_params.get("message", "Broadcast!")
server.push_message(message) # 不指定client_id则广播
return {"status": "broadcasted"}
```
### 中间件参数
中间件可以通过process_request和process_response方法处理请求和响应：
```python
class CustomMiddleware(Middleware):
def __init__(self, **options):
self.options = options
def process_request(self, request):
# 处理请求
request.custom_attr = self.options.get("custom_attr")
return None
def process_response(self, request, response):
# 处理响应
if isinstance(response, dict):
response["processed_by"] = self.options.get("name", "CustomMiddleware")
return response
# 使用中间件
server = Server()
server.middleware_manager.add_middleware(
CustomMiddleware(
name="MyMiddleware",
custom_attr="custom_value"
)
)
```
### 后台任务参数
后台任务支持参数传递：
```python
@Route.register("/task", method="GET")
def async_task(request):
def background_job(user_id, task_type="default"):
time.sleep(5)
Logger.info(f"Processing {task_type} task for user {user_id}")
# 添加带参数的后台任务
user_id = request.query_params.get("user_id", "0")
request.bg.add(
background_job,
user_id, # 位置参数
task_type="important" # 关键字参数
)
return {"message": "Task scheduled"}
```
这些参数的使用示例展示了框架的灵活性和功能性。你可以根据需要组合使用这些参数来实现各种功能。所有参数都有合理的默认值，使框架既易于使用又足够灵活。
需要注意的是：
1. 参数名称区分大小写
2. 必填参数没有默认值时必须提供
3. 某些参数组合可能会相互影响
4. 建议遵循Python的命名规范
## 注意事项
1. TCP长连接需要显式启用：`server.config.tcp_keep_alive = True`
2. 后台任务会在独立的线程池中
3. 推送消息时如果不指定client_id，将广播给所有连接的客户端
## 许可证
MIT License
Raw data
{
"_id": null,
"home_page": "https://github.com/xiwen-haochi/pyliteflux",
"name": "pyliteflux",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": "web framework, tcp server, http server, pure python, learning, teaching, standard library",
"author": "'fkl'",
"author_email": "18654198191@163.com",
"download_url": "https://files.pythonhosted.org/packages/00/a1/ab86ceefbe12d434d9d403ed086075b11e704d887c511f84f7d1f54a5c30/pyliteflux-0.1.2.tar.gz",
"platform": null,
"description": "# PyLiteFlux\n\nPyLiteFlux \u662f\u4e00\u4e2a\u8f7b\u91cf\u7ea7\u7684Python Web/TCP\u670d\u52a1\u5668\u6846\u67b6\uff0c\u5b8c\u5168\u57fa\u4e8ePython\u6807\u51c6\u5e93\u5b9e\u73b0\uff0c\u4e0d\u4f9d\u8d56\u4efb\u4f55\u7b2c\u4e09\u65b9\u5305\u3002\u5b83\u652f\u6301HTTP\u548cTCP\u534f\u8bae\uff0c\u63d0\u4f9b\u4e86\u7b80\u5355\u800c\u5f3a\u5927\u7684API\u6765\u6784\u5efa\u7f51\u7edc\u5e94\u7528\u3002\u9002\u5408\u7528\u4e8e\u5b66\u4e60\u548c\u7406\u89e3Python\u540e\u7aef\u6846\u67b6\u7684\u6982\u5ff5\u548c\u5b9e\u73b0\u539f\u7406\u3002\n\n## \u7279\u6027\n\n- \u652f\u6301HTTP\u548cTCP\u670d\u52a1\u5668\n- \u652f\u6301TCP\u957f\u8fde\u63a5\u548c\u6d88\u606f\u63a8\u9001\n- \u652f\u6301\u540e\u53f0\u4efb\u52a1\u5904\u7406\n- \u5185\u7f6e\u4e2d\u95f4\u4ef6\u7cfb\u7edf\n- \u7b80\u5355\u7684\u8def\u7531\u6ce8\u518c\n- \u652f\u6301\u914d\u7f6e\u7ba1\u7406\n- \u652f\u6301\u8bf7\u6c42\u4e0a\u4e0b\u6587\n\n## \u5b89\u88c5\n\n```bash\npip install pyliteflux\n```\n\n## \u5feb\u901f\u5f00\u59cb\n\n### \u57fa\u672c\u4f7f\u7528\n\n```python\nfrom pyliteflux import Server, Route\n\nserver = Server()\n\n@Route.register(\"/hello\", method=\"GET\")\ndef hello(request):\n return {\"message\": \"Hello, World!\"}\n\nserver.run()\n```\n\n### TCP\u957f\u8fde\u63a5\u548c\u6d88\u606f\u63a8\u9001\u793a\u4f8b\n\n\u4e0b\u9762\u662f\u4e00\u4e2a\u7b80\u5355\u7684\u804a\u5929\u7cfb\u7edf\u793a\u4f8b\uff0c\u5c55\u793a\u4e86\u5982\u4f55\u4f7f\u7528TCP\u957f\u8fde\u63a5\u548c\u6d88\u606f\u63a8\u9001\u529f\u80fd\uff1a\n\n```python\nfrom pyliteflux import Server, Route\nimport time\n\nserver = Server()\nserver.config.tcp_keep_alive = True # \u542f\u7528TCP\u957f\u8fde\u63a5\n\n# HTTP\u63a5\u53e3\uff1a\u53d1\u9001\u6d88\u606f\u7ed9\u6307\u5b9a\u5ba2\u6237\u7aef\n@Route.register(\"/hello\", method=\"GET\")\ndef hello(request):\n # \u4ece\u67e5\u8be2\u53c2\u6570\u4e2d\u83b7\u53d6\u76ee\u6807client_id\n target_client = request.query_params.get(\"client_id\")\n \n def heavy_task():\n time.sleep(5) # \u6a21\u62df\u8017\u65f6\u64cd\u4f5c\n server.push_message(\"Hello via HTTP GET!\", target_client)\n \n request.bg.add(heavy_task)\n return {\n \"message\": \"Hello via HTTP GET!\",\n \"target_client\": target_client,\n \"query_params\": request.query_params\n }\n\n# TCP\u63a5\u53e3\uff1a\u5217\u51fa\u6240\u6709\u6d3b\u8dc3\u8fde\u63a5\n@Route.register(\"clients\", protocol=\"tcp\")\ndef list_clients(request):\n clients = list(server._active_connections.keys())\n return {\n \"message\": \"Active TCP clients\",\n \"clients\": clients,\n \"count\": len(clients)\n }\n\n# TCP\u63a5\u53e3\uff1a\u8ba2\u9605\u6d88\u606f\n@Route.register(\"subscribe\", protocol=\"tcp\")\ndef subscribe_tcp(request):\n client_id = f\"{request.client_address[0]}:{request.client_address[1]}\"\n return {\"message\": \"subscribed\", \"client_id\": client_id}\n\nserver.run()\n```\n\n## \u4e3b\u8981\u529f\u80fd\n\n### HTTP\u670d\u52a1\u5668\n- \u652f\u6301GET\u3001POST\u3001PUT\u3001DELETE\u65b9\u6cd5\n- \u652f\u6301\u8def\u7531\u53c2\u6570\n- \u652f\u6301\u67e5\u8be2\u53c2\u6570\n- \u652f\u6301JSON\u548c\u8868\u5355\u6570\u636e\n\n### TCP\u670d\u52a1\u5668\n- \u652f\u6301\u957f\u8fde\u63a5\n- \u652f\u6301\u6d88\u606f\u63a8\u9001\n- \u652f\u6301JSON\u548cXML\u6570\u636e\u683c\u5f0f\n- \u652f\u6301\u81ea\u5b9a\u4e49\u547d\u4ee4\n\n### \u540e\u53f0\u4efb\u52a1\n- \u652f\u6301\u5f02\u6b65\u4efb\u52a1\u5904\u7406\n- \u4efb\u52a1\u961f\u5217\u7ba1\u7406\n- \u7ebf\u7a0b\u6c60\u6267\u884c\n\n### \u4e2d\u95f4\u4ef6\u7cfb\u7edf\n- \u8bf7\u6c42\u5904\u7406\u4e2d\u95f4\u4ef6\n- \u54cd\u5e94\u5904\u7406\u4e2d\u95f4\u4ef6\n- \u4e0a\u4e0b\u6587\u7ba1\u7406\u4e2d\u95f4\u4ef6\n\n### \u4e2d\u95f4\u4ef6\u548c\u5168\u5c40\u4e0a\u4e0b\u6587\u793a\u4f8b\n\n#### \u4e2d\u95f4\u4ef6\u4f7f\u7528\n\n\u4e2d\u95f4\u4ef6\u53ef\u4ee5\u5904\u7406\u8bf7\u6c42\u548c\u54cd\u5e94\uff0c\u5b9e\u73b0\u8ba4\u8bc1\u3001\u65e5\u5fd7\u3001\u6027\u80fd\u76d1\u63a7\u7b49\u529f\u80fd\uff1a\n\n```python\nfrom pyliteflux import Server, Route, Middleware\n\n# \u5b9a\u4e49\u8ba4\u8bc1\u4e2d\u95f4\u4ef6\nclass AuthMiddleware(Middleware):\n def process_request(self, request):\n # \u68c0\u67e5\u8ba4\u8bc1token\n token = request.headers.get(\"Authorization\")\n if not token:\n return {\"error\": \"Unauthorized\"}\n \n # \u5c06\u8ba4\u8bc1\u4fe1\u606f\u5b58\u50a8\u5728g\u5bf9\u8c61\u4e2d\n g.user = self.validate_token(token)\n return None # \u7ee7\u7eed\u5904\u7406\u8bf7\u6c42\n \n def process_response(self, request, response):\n # \u5904\u7406\u54cd\u5e94\n if isinstance(response, dict):\n response[\"authenticated\"] = hasattr(g, \"user\")\n return response\n \n def validate_token(self, token):\n # \u5b9e\u9645\u9879\u76ee\u4e2d\u8fd9\u91cc\u5e94\u8be5\u9a8c\u8bc1token\n return {\"id\": 1, \"name\": \"user1\"}\n\n# \u5b9a\u4e49\u65e5\u5fd7\u4e2d\u95f4\u4ef6\nclass LogMiddleware(Middleware):\n def process_request(self, request):\n # \u8bb0\u5f55\u8bf7\u6c42\u5f00\u59cb\u65f6\u95f4\n g.start_time = time.time()\n return None\n \n def process_response(self, request, response):\n # \u8ba1\u7b97\u8bf7\u6c42\u5904\u7406\u65f6\u95f4\n duration = time.time() - g.start_time\n Logger.info(f\"Request processed in {duration:.2f}s\")\n return response\n\n# \u4f7f\u7528\u4e2d\u95f4\u4ef6\nserver = Server()\nserver.middleware_manager.add_middleware(AuthMiddleware())\nserver.middleware_manager.add_middleware(LogMiddleware())\n\n# \u5728\u8def\u7531\u4e2d\u4f7f\u7528g\u5bf9\u8c61\n@Route.register(\"/user\", method=\"GET\")\ndef get_user(request):\n if hasattr(g, \"user\"):\n return {\n \"message\": \"User info\",\n \"user\": g.user,\n \"request_id\": g.request_id # \u6bcf\u4e2a\u8bf7\u6c42\u7684\u552f\u4e00ID\n }\n return {\"error\": \"User not found\"}\n\nserver.run()\n```\n\n#### \u5168\u5c40\u4e0a\u4e0b\u6587\u4f7f\u7528\n\ng\u5bf9\u8c61\u662f\u8bf7\u6c42\u7ea7\u522b\u7684\u5168\u5c40\u5bf9\u8c61\uff0c\u53ef\u4ee5\u5728\u6574\u4e2a\u8bf7\u6c42\u8fc7\u7a0b\u4e2d\u5171\u4eab\u6570\u636e\uff1a\n\n```python\nfrom pyliteflux import Server, Route, g\n\nserver = Server()\n\n# \u5728\u4e2d\u95f4\u4ef6\u4e2d\u8bbe\u7f6e\u6570\u636e\nclass DataMiddleware(Middleware):\n def process_request(self, request):\n g.db = Database() # \u5047\u8bbe\u7684\u6570\u636e\u5e93\u8fde\u63a5\n g.cache = Cache() # \u5047\u8bbe\u7684\u7f13\u5b58\u8fde\u63a5\n return None\n \n def process_response(self, request, response):\n # \u6e05\u7406\u8d44\u6e90\n if hasattr(g, \"db\"):\n g.db.close()\n if hasattr(g, \"cache\"):\n g.cache.close()\n return response\n\n# \u5728\u8def\u7531\u4e2d\u4f7f\u7528g\u5bf9\u8c61\n@Route.register(\"/data\", method=\"GET\")\ndef get_data(request):\n # \u4f7f\u7528g\u5bf9\u8c61\u4e2d\u7684\u8d44\u6e90\n data = g.db.query(\"SELECT * FROM users\")\n cached_data = g.cache.get(\"users\")\n \n return {\n \"data\": data,\n \"cached\": cached_data,\n \"request_id\": g.request_id\n }\n\n# \u5728\u540e\u53f0\u4efb\u52a1\u4e2d\u4f7f\u7528g\u5bf9\u8c61\n@Route.register(\"/async\", method=\"GET\")\ndef async_task(request):\n def background_job():\n # \u6ce8\u610f\uff1a\u540e\u53f0\u4efb\u52a1\u6709\u81ea\u5df1\u7684g\u5bf9\u8c61\u5b9e\u4f8b\n g.task_id = uuid.uuid4()\n Logger.info(f\"Background task {g.task_id} started\")\n \n request.bg.add(background_job)\n return {\"message\": \"Task scheduled\"}\n\nserver.run()\n```\n\n#### \u5b9e\u9645\u5e94\u7528\u573a\u666f\n\n1. \u7528\u6237\u8ba4\u8bc1\u548c\u6388\u6743\uff1a\n```python\nclass AuthMiddleware(Middleware):\n def process_request(self, request):\n token = request.headers.get(\"Authorization\")\n if token:\n g.user = self.authenticate(token)\n g.permissions = self.get_permissions(g.user)\n return None\n\n@Route.register(\"/admin\", method=\"GET\")\ndef admin_panel(request):\n if not hasattr(g, \"permissions\") or \"admin\" not in g.permissions:\n return {\"error\": \"Access denied\"}, 403\n return {\"message\": \"Welcome, admin!\"}\n```\n\n2. \u6027\u80fd\u76d1\u63a7\uff1a\n```python\nclass PerformanceMiddleware(Middleware):\n def process_request(self, request):\n g.start_time = time.time()\n g.sql_queries = []\n return None\n \n def process_response(self, request, response):\n duration = time.time() - g.start_time\n metrics = {\n \"duration\": duration,\n \"sql_queries\": len(g.sql_queries),\n \"memory_usage\": self.get_memory_usage()\n }\n Logger.info(f\"Performance metrics: {metrics}\")\n return response\n```\n\n3. \u8bf7\u6c42\u8ddf\u8e2a\uff1a\n```python\nclass TracingMiddleware(Middleware):\n def process_request(self, request):\n g.trace_id = str(uuid.uuid4())\n g.span_id = str(uuid.uuid4())\n g.traces = []\n return None\n \n def process_response(self, request, response):\n if isinstance(response, dict):\n response[\"trace_id\"] = g.trace_id\n response[\"traces\"] = g.traces\n return response\n\n@Route.register(\"/trace\", method=\"GET\")\ndef traced_endpoint(request):\n g.traces.append({\"event\": \"processing\", \"time\": time.time()})\n result = process_data()\n g.traces.append({\"event\": \"completed\", \"time\": time.time()})\n return {\"data\": result}\n```\n\n\u8fd9\u4e9b\u793a\u4f8b\u5c55\u793a\u4e86\u4e2d\u95f4\u4ef6\u548cg\u5bf9\u8c61\u7684\u5f3a\u5927\u529f\u80fd\uff1a\n1. \u8bf7\u6c42\u524d\u540e\u7684\u5904\u7406\n2. \u8d44\u6e90\u7684\u81ea\u52a8\u7ba1\u7406\n3. \u8bf7\u6c42\u7ea7\u522b\u7684\u6570\u636e\u5171\u4eab\n4. \u6027\u80fd\u76d1\u63a7\u548c\u65e5\u5fd7\u8bb0\u5f55\n5. \u8ba4\u8bc1\u548c\u6388\u6743\n6. \u8bf7\u6c42\u8ddf\u8e2a\u548c\u8c03\u8bd5\n\n\u4e2d\u95f4\u4ef6\u548cg\u5bf9\u8c61\u7684\u7ec4\u5408\u4f7f\u7528\u53ef\u4ee5\u5927\u5927\u63d0\u9ad8\u5f00\u53d1\u6548\u7387\u548c\u4ee3\u7801\u8d28\u91cf\u3002\n\n## \u914d\u7f6e\u9009\u9879\n\n```python\nserver = Server()\nserver.config.update(\n http_host=\"127.0.0.1\",\n http_port=8000,\n tcp_host=\"127.0.0.1\",\n tcp_port=8001,\n tcp_keep_alive=True, # \u542f\u7528TCP\u957f\u8fde\u63a5\n info=True # \u542f\u7528\u65e5\u5fd7\n)\n```\n\n## API\u6587\u6863\n\n### \u8def\u7531\u88c5\u9970\u5668\n```python\n@Route.register(path, method=\"GET\", protocol=\"http\")\n```\n- path: \u8def\u7531\u8def\u5f84\n- method: HTTP\u65b9\u6cd5\uff08GET/POST/PUT/DELETE\uff09\n- protocol: \u534f\u8bae\u7c7b\u578b\uff08http/tcp\uff09\n\n### \u670d\u52a1\u5668\u65b9\u6cd5\n```python\nserver.run() # \u542f\u52a8\u670d\u52a1\u5668\nserver.stop() # \u505c\u6b62\u670d\u52a1\u5668\nserver.push_message(message, client_id=None) # \u63a8\u9001\u6d88\u606f\n```\n\n### \u8bf7\u6c42\u5bf9\u8c61\n```python\nrequest.query_params # \u67e5\u8be2\u53c2\u6570\nrequest.form_data # \u8868\u5355\u6570\u636e\nrequest.json_data # JSON\u6570\u636e\nrequest.headers # \u8bf7\u6c42\u5934\nrequest.bg.add(task) # \u6dfb\u52a0\u540e\u53f0\u4efb\u52a1\n```\n\n### \u54cd\u5e94\u683c\u5f0f\n\nPyLiteFlux\u652f\u6301\u591a\u79cd\u54cd\u5e94\u683c\u5f0f\uff0c\u5305\u62ec\u666e\u901aJSON\u54cd\u5e94\u3001\u6d41\u5f0f\u54cd\u5e94\u548c\u81ea\u5b9a\u4e49JSON\u54cd\u5e94\uff1a\n\n#### 1. \u666e\u901aJSON\u54cd\u5e94\n\n````python\n@Route.register(\"/hello\", method=\"GET\")\ndef hello(request):\n return {\"message\": \"Hello, World!\"} # \u81ea\u52a8\u8f6c\u6362\u4e3aJSON\u54cd\u5e94\n````\n\n#### 2. \u6d41\u5f0f\u54cd\u5e94\n\n\u6d41\u5f0f\u54cd\u5e94\u9002\u7528\u4e8e\u5927\u6570\u636e\u4f20\u8f93\u6216\u9700\u8981\u5b9e\u65f6\u63a8\u9001\u6570\u636e\u7684\u573a\u666f\uff1a\n\n````python\nfrom pyliteflux import StreamResponse\n\n@Route.register(\"/stream\", method=\"GET\")\ndef stream_data(request):\n def generate_data():\n for i in range(10):\n time.sleep(1) # \u6a21\u62df\u6570\u636e\u751f\u6210\n yield f\"data: {i}\\n\\n\"\n \n return StreamResponse(\n generate_data(),\n content_type=\"text/event-stream\",\n headers={\n \"Cache-Control\": \"no-cache\",\n \"Connection\": \"keep-alive\"\n }\n )\n\n# \u5e26\u53c2\u6570\u7684\u6d41\u5f0f\u54cd\u5e94\n@Route.register(\"/download\", method=\"GET\")\ndef download_file(request):\n def file_reader():\n with open(\"large_file.txt\", \"rb\") as f:\n while chunk := f.read(8192):\n yield chunk\n \n return StreamResponse(\n file_reader(),\n content_type=\"application/octet-stream\",\n headers={\n \"Content-Disposition\": \"attachment; filename=large_file.txt\"\n }\n )\n````\n\n#### 3. HTTPJSONResponse\u54cd\u5e94\n\n\u5f53\u9700\u8981\u66f4\u7ec6\u7c92\u5ea6\u5730\u63a7\u5236JSON\u54cd\u5e94\u65f6\uff0c\u53ef\u4ee5\u4f7f\u7528HTTPJSONResponse\uff1a\n\n````python\nfrom pyliteflux import HTTPJSONResponse\n\n@Route.register(\"/api/user\", method=\"GET\")\ndef get_user(request):\n user_data = {\"id\": 1, \"name\": \"John\"}\n return HTTPJSONResponse(\n data=user_data,\n status=200,\n headers={\n \"X-Custom-Header\": \"value\",\n \"Access-Control-Allow-Origin\": \"*\"\n }\n )\n\n# \u9519\u8bef\u54cd\u5e94\u793a\u4f8b\n@Route.register(\"/api/error\", method=\"GET\")\ndef error_response(request):\n return HTTPJSONResponse(\n data={\"error\": \"Not Found\", \"code\": \"404\"},\n status=404,\n headers={\"X-Error-Type\": \"NotFound\"}\n )\n\n# \u81ea\u5b9a\u4e49\u5e8f\u5217\u5316\u793a\u4f8b\n@Route.register(\"/api/custom\", method=\"GET\")\ndef custom_serialization(request):\n class CustomEncoder(json.JSONEncoder):\n def default(self, obj):\n if isinstance(obj, datetime):\n return obj.isoformat()\n return super().default(obj)\n \n data = {\n \"timestamp\": datetime.now(),\n \"message\": \"Custom serialization\"\n }\n \n return HTTPJSONResponse(\n data=data,\n json_encoder=CustomEncoder,\n headers={\"Content-Type\": \"application/json\"}\n )\n````\n\n#### 4. \u7ec4\u5408\u4f7f\u7528\u793a\u4f8b\n\n````python\n@Route.register(\"/api/complex\", method=\"GET\")\ndef complex_response(request):\n response_type = request.query_params.get(\"type\", \"json\")\n \n if response_type == \"stream\":\n def stream():\n for i in range(5):\n yield json.dumps({\"count\": i}) + \"\\n\"\n return StreamResponse(stream(), content_type=\"application/x-ndjson\")\n \n elif response_type == \"error\":\n return HTTPJSONResponse(\n {\"error\": \"Bad Request\"},\n status=400,\n headers={\"X-Error-Details\": \"Invalid type\"}\n )\n \n else:\n return {\"message\": \"Regular JSON response\"} # \u81ea\u52a8\u8f6c\u6362\u4e3aJSON\n````\n\n\n\n## \u53c2\u6570\u4f7f\u7528\u8bf4\u660e\n\n### \u670d\u52a1\u5668\u914d\u7f6e\u53c2\u6570\n\n\u670d\u52a1\u5668\u5b9e\u4f8b\u652f\u6301\u591a\u79cd\u914d\u7f6e\u53c2\u6570\uff0c\u53ef\u4ee5\u901a\u8fc7config\u5bf9\u8c61\u8fdb\u884c\u8bbe\u7f6e\uff1a\n\n```python\nfrom pyliteflux import Server\n\nserver = Server()\n\n# \u65b9\u5f0f1\uff1a\u901a\u8fc7\u5c5e\u6027\u8bbe\u7f6e\nserver.config.http_host = \"127.0.0.1\"\nserver.config.http_port = 8000\nserver.config.tcp_keep_alive = True\n\n# \u65b9\u5f0f2\uff1a\u901a\u8fc7update\u65b9\u6cd5\u6279\u91cf\u8bbe\u7f6e\nserver.config.update(\n http_host=\"127.0.0.1\",\n http_port=8000,\n tcp_host=\"127.0.0.1\",\n tcp_port=8001,\n tcp_keep_alive=True,\n info=True\n)\n```\n\n\u914d\u7f6e\u53c2\u6570\u8bf4\u660e\uff1a\n| \u53c2\u6570 | \u7c7b\u578b | \u9ed8\u8ba4\u503c | \u8bf4\u660e |\n|------|------|--------|------|\n| http_host | str | \"127.0.0.1\" | HTTP\u670d\u52a1\u5668\u4e3b\u673a\u5730\u5740 |\n| http_port | int | 8000 | HTTP\u670d\u52a1\u5668\u7aef\u53e3 |\n| tcp_host | str | \"127.0.0.1\" | TCP\u670d\u52a1\u5668\u4e3b\u673a\u5730\u5740 |\n| tcp_port | int | 8001 | TCP\u670d\u52a1\u5668\u7aef\u53e3 |\n| tcp_keep_alive | bool | False | \u662f\u5426\u542f\u7528TCP\u957f\u8fde\u63a5 |\n| info | bool | True | \u662f\u5426\u542f\u7528\u65e5\u5fd7\u8f93\u51fa |\n\n### \u8def\u7531\u53c2\u6570\n\n\u8def\u7531\u88c5\u9970\u5668\u652f\u6301\u591a\u79cd\u53c2\u6570\u914d\u7f6e\uff1a\n\n```python\nfrom pyliteflux import Route\n\n# HTTP\u8def\u7531\n@Route.register(\"/hello\", method=\"GET\") # \u57fa\u672cGET\u8bf7\u6c42\n@Route.register(\"/user\", method=\"POST\") # POST\u8bf7\u6c42\n@Route.register(\"/user/{id}\", method=\"GET\") # \u5e26\u8def\u5f84\u53c2\u6570\u7684\u8def\u7531\n@Route.register(\"/files/*\", method=\"GET\") # \u901a\u914d\u7b26\u8def\u7531\n\n# TCP\u8def\u7531\n@Route.register(\"command\", protocol=\"tcp\") # TCP\u547d\u4ee4\u8def\u7531\n```\n\n\u8def\u7531\u53c2\u6570\u8bf4\u660e\uff1a\n| \u53c2\u6570 | \u7c7b\u578b | \u9ed8\u8ba4\u503c | \u8bf4\u660e |\n|------|------|--------|------|\n| path | str | \u5fc5\u586b | \u8def\u7531\u8def\u5f84 |\n| method | str | \"GET\" | HTTP\u65b9\u6cd5\uff08GET/POST/PUT/DELETE\uff09 |\n| protocol | str | \"http\" | \u534f\u8bae\u7c7b\u578b\uff08http/tcp\uff09 |\n\n### \u8bf7\u6c42\u53c2\u6570\n\nHTTP\u8bf7\u6c42\u53ef\u4ee5\u901a\u8fc7\u591a\u79cd\u65b9\u5f0f\u4f20\u9012\u53c2\u6570\uff1a\n\n```python\n# 1. \u67e5\u8be2\u53c2\u6570\n@Route.register(\"/search\", method=\"GET\")\ndef search(request):\n keyword = request.query_params.get(\"keyword\")\n page = request.query_params.get(\"page\", \"1\")\n return {\"keyword\": keyword, \"page\": page}\n\n# 2. \u8def\u5f84\u53c2\u6570\n@Route.register(\"/user/{user_id}/posts/{post_id}\", method=\"GET\")\ndef get_post(request):\n user_id = request.path_params[\"user_id\"]\n post_id = request.path_params[\"post_id\"]\n return {\"user_id\": user_id, \"post_id\": post_id}\n\n# 3. \u8868\u5355\u6570\u636e\n@Route.register(\"/upload\", method=\"POST\")\ndef upload(request):\n file_name = request.form_data.get(\"file_name\")\n content = request.form_data.get(\"content\")\n return {\"file_name\": file_name}\n\n# 4. JSON\u6570\u636e\n@Route.register(\"/api/data\", method=\"POST\")\ndef handle_json(request):\n data = request.json_data\n return {\"received\": data}\n```\n\n### TCP\u8fde\u63a5\u53c2\u6570\n\nTCP\u8fde\u63a5\u76f8\u5173\u7684\u53c2\u6570\u548c\u4f7f\u7528\u65b9\u5f0f\uff1a\n\n```python\n# 1. \u542f\u7528TCP\u957f\u8fde\u63a5\nserver = Server()\nserver.config.tcp_keep_alive = True\n\n# 2. \u53d1\u9001\u6d88\u606f\u7ed9\u7279\u5b9a\u5ba2\u6237\u7aef\n@Route.register(\"/push\", method=\"GET\")\ndef push_message(request):\n client_id = request.query_params.get(\"client_id\")\n message = request.query_params.get(\"message\", \"Hello!\")\n server.push_message(message, client_id)\n return {\"status\": \"sent\"}\n\n# 3. \u5e7f\u64ad\u6d88\u606f\u7ed9\u6240\u6709\u5ba2\u6237\u7aef\n@Route.register(\"/broadcast\", method=\"GET\")\ndef broadcast(request):\n message = request.query_params.get(\"message\", \"Broadcast!\")\n server.push_message(message) # \u4e0d\u6307\u5b9aclient_id\u5219\u5e7f\u64ad\n return {\"status\": \"broadcasted\"}\n```\n\n### \u4e2d\u95f4\u4ef6\u53c2\u6570\n\n\u4e2d\u95f4\u4ef6\u53ef\u4ee5\u901a\u8fc7process_request\u548cprocess_response\u65b9\u6cd5\u5904\u7406\u8bf7\u6c42\u548c\u54cd\u5e94\uff1a\n\n```python\nclass CustomMiddleware(Middleware):\n def __init__(self, **options):\n self.options = options\n \n def process_request(self, request):\n # \u5904\u7406\u8bf7\u6c42\n request.custom_attr = self.options.get(\"custom_attr\")\n return None\n \n def process_response(self, request, response):\n # \u5904\u7406\u54cd\u5e94\n if isinstance(response, dict):\n response[\"processed_by\"] = self.options.get(\"name\", \"CustomMiddleware\")\n return response\n\n# \u4f7f\u7528\u4e2d\u95f4\u4ef6\nserver = Server()\nserver.middleware_manager.add_middleware(\n CustomMiddleware(\n name=\"MyMiddleware\",\n custom_attr=\"custom_value\"\n )\n)\n```\n\n### \u540e\u53f0\u4efb\u52a1\u53c2\u6570\n\n\u540e\u53f0\u4efb\u52a1\u652f\u6301\u53c2\u6570\u4f20\u9012\uff1a\n\n```python\n@Route.register(\"/task\", method=\"GET\")\ndef async_task(request):\n def background_job(user_id, task_type=\"default\"):\n time.sleep(5)\n Logger.info(f\"Processing {task_type} task for user {user_id}\")\n \n # \u6dfb\u52a0\u5e26\u53c2\u6570\u7684\u540e\u53f0\u4efb\u52a1\n user_id = request.query_params.get(\"user_id\", \"0\")\n request.bg.add(\n background_job,\n user_id, # \u4f4d\u7f6e\u53c2\u6570\n task_type=\"important\" # \u5173\u952e\u5b57\u53c2\u6570\n )\n \n return {\"message\": \"Task scheduled\"}\n```\n\n\u8fd9\u4e9b\u53c2\u6570\u7684\u4f7f\u7528\u793a\u4f8b\u5c55\u793a\u4e86\u6846\u67b6\u7684\u7075\u6d3b\u6027\u548c\u529f\u80fd\u6027\u3002\u4f60\u53ef\u4ee5\u6839\u636e\u9700\u8981\u7ec4\u5408\u4f7f\u7528\u8fd9\u4e9b\u53c2\u6570\u6765\u5b9e\u73b0\u5404\u79cd\u529f\u80fd\u3002\u6240\u6709\u53c2\u6570\u90fd\u6709\u5408\u7406\u7684\u9ed8\u8ba4\u503c\uff0c\u4f7f\u6846\u67b6\u65e2\u6613\u4e8e\u4f7f\u7528\u53c8\u8db3\u591f\u7075\u6d3b\u3002\n\n\u9700\u8981\u6ce8\u610f\u7684\u662f\uff1a\n1. \u53c2\u6570\u540d\u79f0\u533a\u5206\u5927\u5c0f\u5199\n2. \u5fc5\u586b\u53c2\u6570\u6ca1\u6709\u9ed8\u8ba4\u503c\u65f6\u5fc5\u987b\u63d0\u4f9b\n3. \u67d0\u4e9b\u53c2\u6570\u7ec4\u5408\u53ef\u80fd\u4f1a\u76f8\u4e92\u5f71\u54cd\n4. \u5efa\u8bae\u9075\u5faaPython\u7684\u547d\u540d\u89c4\u8303\n\n\n## \u6ce8\u610f\u4e8b\u9879\n\n1. TCP\u957f\u8fde\u63a5\u9700\u8981\u663e\u5f0f\u542f\u7528\uff1a`server.config.tcp_keep_alive = True`\n2. \u540e\u53f0\u4efb\u52a1\u4f1a\u5728\u72ec\u7acb\u7684\u7ebf\u7a0b\u6c60\u4e2d\n3. \u63a8\u9001\u6d88\u606f\u65f6\u5982\u679c\u4e0d\u6307\u5b9aclient_id\uff0c\u5c06\u5e7f\u64ad\u7ed9\u6240\u6709\u8fde\u63a5\u7684\u5ba2\u6237\u7aef\n\n## \u8bb8\u53ef\u8bc1\n\nMIT License",
"bugtrack_url": null,
"license": "MIT",
"summary": "A lightweight Python Web/TCP server framework implemented with pure Python standard library",
"version": "0.1.2",
"project_urls": {
"Homepage": "https://github.com/xiwen-haochi/pyliteflux",
"Repository": "https://github.com/xiwen-haochi/pyliteflux"
},
"split_keywords": [
"web framework",
" tcp server",
" http server",
" pure python",
" learning",
" teaching",
" standard library"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "2ef2f8f11f16e74f4b3f889516fa91279f766e08681297ce6c2dae9ea25692ce",
"md5": "517d9a044ab1e8b683bb5f3a0ceb3bc8",
"sha256": "16f56e0ac8a13d6182ef91c584fe9f3c0750efeb81529f607b63aa0559621673"
},
"downloads": -1,
"filename": "pyliteflux-0.1.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "517d9a044ab1e8b683bb5f3a0ceb3bc8",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 16021,
"upload_time": "2024-11-27T03:41:38",
"upload_time_iso_8601": "2024-11-27T03:41:38.919031Z",
"url": "https://files.pythonhosted.org/packages/2e/f2/f8f11f16e74f4b3f889516fa91279f766e08681297ce6c2dae9ea25692ce/pyliteflux-0.1.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "00a1ab86ceefbe12d434d9d403ed086075b11e704d887c511f84f7d1f54a5c30",
"md5": "bdc1a241f633ceb8b831846f5ed448a1",
"sha256": "e40b9de74711ee7cf12e42ce1c70d5637c7b95b4313f0636e1782a82bc1c1a6a"
},
"downloads": -1,
"filename": "pyliteflux-0.1.2.tar.gz",
"has_sig": false,
"md5_digest": "bdc1a241f633ceb8b831846f5ed448a1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 16976,
"upload_time": "2024-11-27T03:41:40",
"upload_time_iso_8601": "2024-11-27T03:41:40.801135Z",
"url": "https://files.pythonhosted.org/packages/00/a1/ab86ceefbe12d434d9d403ed086075b11e704d887c511f84f7d1f54a5c30/pyliteflux-0.1.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-11-27 03:41:40",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "xiwen-haochi",
"github_project": "pyliteflux",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "pyliteflux"
}
JSON
