Python使用socket实现简易的http服务

在接触的一些项目中,有时为了方便可视化一些服务状态(请求数很少),那么很容易想到使用http服务来实现。但开源的web后端框架,例如flask,fastapi,django等略显沉重,且使用这些框架会有各种各样的限制,为了更加灵活的使用,可以自己通过Python自带的socket库来实现,下面是我简单实现的代码,贴在这里方便后续使用。

客户端代码:

import os
import time
import json
import socket
import datetime
import traceback
from urllib import parseENCODING = "utf-8"class CustomRouter:def __init__(self):self.router_dict = {}def register(self, url: str, method: str):if method not in ["GET", "POST"]:raise ValueError(f"method only support GET or POST, got {method}")key = f"{url}:{method}"if key in self.router_dict:raise ValueError(f"url:{url} method:{method} already registed")def _register(func):self.router_dict[key] = funcreturn funcreturn _registerdef has_register(self, url: str, method: str):key = f"{url}:{method}"return key in self.router_dictdef request(self, url: str, method: str):key = f"{url}:{method}"func = self.router_dict[key]return func()router = CustomRouter()@router.register(url="/", method="GET")
def task():return "Hello World!!!"@router.register(url="/get_project_info", method="GET")
def get_project_info():info = {"pwd": os.path.abspath(os.curdir),"encoding": ENCODING,"time": datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}return json.dumps(info)class CustomServer:def __init__(self, port: int = 5055, backlog: int = 2):self.server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server.bind(("", port))self.server.listen(backlog)self.buff_size = 1024self.encoding = ENCODINGself.router = routerself.timeout = 1.0def parse_message(self, message: str):method, url_params, message = message.split(" ", maxsplit=2)res = parse.urlparse(url_params)url = res.pathparams = dict([i for i in parse.parse_qsl(res.query)])version, message = message.split("\r\n", maxsplit=1)return method, url, params, versiondef run(self):while True:conn, addr = self.server.accept()# conn.settimeout(self.timeout)with conn:try:message = conn.recv(self.buff_size).decode(self.encoding)t0 = time.perf_counter()method, url, params, version = self.parse_message(message)if not self.router.has_register(url, method):conn.send("HTTP/1.1 501 Not Implemented\r\n\r\n".encode(self.encoding))print(f"[{method}][501]client addr: {addr} request: {url}, time: {time.perf_counter() - t0:.5f}")continueres = self.router.request(url, method)conn.send("HTTP/1.1 200 OK\r\n\r\n".encode(self.encoding))conn.send(res.encode(self.encoding))print(f"[{method}][200]client addr: {addr} request: {url}, time: {time.perf_counter() - t0:.5f}")except (BrokenPipeError, TimeoutError):print(traceback.format_exc())except:print(traceback.format_exc())conn.send("HTTP/1.1 500 Inner Error\r\n\r\n".encode(self.encoding))print(f"[{method}][500]client addr: {addr} request: {url}, time: {time.perf_counter() - t0:.5f}")if __name__ == '__main__':s = CustomServer()s.run()

客户端代码:

import requestsdef main():for _ in range(50):res = requests.get("http://127.0.0.1:5055")print(res.status_code, res.text)res = requests.get("http://127.0.0.1:5055/get_project_info")print(res.status_code, res.json())if __name__ == '__main__':main()

客户端请求时,服务端终端输出:

[GET][200]client addr: ('127.0.0.1', 38054) request: /, time: 0.00002
[GET][200]client addr: ('127.0.0.1', 38058) request: /get_project_info, time: 0.00004
[GET][200]client addr: ('127.0.0.1', 38060) request: /, time: 0.00002
[GET][200]client addr: ('127.0.0.1', 38070) request: /get_project_info, time: 0.00004
[GET][200]client addr: ('127.0.0.1', 38080) request: /, time: 0.00002
[GET][200]client addr: ('127.0.0.1', 38096) request: /get_project_info, time: 0.00004
[GET][200]client addr: ('127.0.0.1', 38098) request: /, time: 0.00002
[GET][200]client addr: ('127.0.0.1', 38114) request: /get_project_info, time: 0.00009
[GET][200]client addr: ('127.0.0.1', 38120) request: /, time: 0.00003
[GET][200]client addr: ('127.0.0.1', 38122) request: /get_project_info, time: 0.00006
[GET][200]client addr: ('127.0.0.1', 38136) request: /, time: 0.00003
[GET][200]client addr: ('127.0.0.1', 38146) request: /get_project_info, time: 0.00006
[GET][200]client addr: ('127.0.0.1', 38160) request: /, time: 0.00003
[GET][200]client addr: ('127.0.0.1', 38172) request: /get_project_info, time: 0.00005
[GET][200]client addr: ('127.0.0.1', 38178) request: /, time: 0.00002
[GET][200]client addr: ('127.0.0.1', 38182) request: /get_project_info, time: 0.00005

客户端终端输出:

200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}
200 Hello World!!!
200 {'pwd': '/home/wz/my_projects/others/socket_t', 'encoding': 'utf-8', 'time': '2025-01-14 00:20:01'}

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mzph.cn/news/892706.shtml

如若内容造成侵权/违法违规/事实不符,请联系多彩编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

25/1/13 嵌入式笔记 继续学习Esp32

PWM(Pulse Width Modulation,脉宽调制) 是一种通过快速切换高低电平来模拟中间电压值的技术。它广泛应用于控制 LED 亮度、电机速度、音频生成等场景。 analogWrite函数:用于在微控制器(如 Arduino)上生成模拟信号。 …

jupyter notebook练手项目:线性回归——学习时间与成绩的关系

线性回归——学习时间与学习成绩的关系 第1步:导入工具库 pandas——数据分析库,提供了数据结构(如DataFrame和Series)和数据操作方法,方便对数据集进行读取、清洗、转换等操作。 matplotlib——绘图库,p…

如何规模化实现完全自动驾驶?Mobileye提出解题“新”思路

在CES 2025上,Mobileye展示了端到端自动驾驶系统Mobileye Drive™,通过高度集成的传感器、算法和计算平台,可以实现自动驾驶功能的全覆盖。 Mobileye创始人兼首席执行官Amnon Shashua教授 期间,Mobileye创始人兼首席执行官Amnon …

Windows下安装和配置Go开发环境

文章目录 1. 介绍了SDK2. 下载 SDK工具包3. windows 下配置 Golang 环境变量 1. 介绍了SDK SDK 的全称(Software Development Kit 软件开发工具包)SDK是提供给开发人员使用的,其中包含了对应开发语言的工具包 2. 下载 SDK工具包 Go语言的官网为:https…

【JVM中的三色标记法是什么?】

JVM中的三色标记法是什么? 一、基本概念二、标记过程三、优势与问题四、漏标与多标的解决方案三色标记法(Tri-color Marking Algorithm)是Java虚拟机(JVM)中一种用于追踪对象存活状态的垃圾回收算法。 它基于William D. Hana和Mark S. McCulleghan在1976年提出的两色标记法…

【javascript】Web APIs-Dom获取属性操作

目录 Web APIs-Dom获取&属性操作 Web API 基本认知 变量声明 建议: const 优先,尽量使用const,原因是: 1.1 作用和分类 1.2 什么是DOM 1.3 DOM树 1.4 DOM对象(重要) 2. 获取DOM对象 2.1 根据C…

新版AndroidStudio通过系统快捷创建带BottomNavigationView的项目踩坑记录

选择上面这个玩意创建的项目 坑点1 :配置的写法和不一样了 镜像的写法: 新的settings.gradle.kts中配置镜像的代码: pluginManagement {repositories {mavenCentral()google {content {includeGroupByRegex("com\\.android.*")…

【0390】Postgres内核 启动 checkpointer process ( 1 )

文章目录 1. signal 间接启动1.1 signal callback 初始化1.2 处理 child process 各种退出状态1.3 start checkpointer process1. signal 间接启动 checkpointer process 的主要入口点是 CheckpointerMain(), 它是从 AuxiliaryProcessMain() 调用的,AuxiliaryProcessMain() …

《AI赋能鸿蒙Next,打造极致沉浸感游戏》

在游戏开发领域,鸿蒙Next系统与人工智能技术的结合为开发者们带来了前所未有的机遇,使打造更具沉浸感的游戏成为可能。以下将深入探讨如何利用人工智能在鸿蒙Next上开发出令人身临其境的游戏。 利用AI优化游戏角色智能行为 在传统游戏中,非…

SAP资产盘盈盘亏的过账处理、入账价值错误调整、资产减值准备

文章目录 一、SAP资产盘盈盘亏处理1、ABNAN盘盈 (往年资产) ABZON (当年资产)2、ABAVN盘亏 二、资产价值入账错了(价值多了或少了),怎么调账1、价值少了2、价值多了 三、资产减值准备1、启用重估2、指定间隔…

C# 迭代,递归,回调--13

目录 一.迭代 迭代器示例: 关键点: 优势: 二.递归 递归示例: 关键点: 优势: 注意: 三.回调 回调示例: 关键点: 优势: 应用场景: 4.三种模式的特点对比: 迭代: 递归: 回调: 一.迭代 在C#中迭代通常指重复执行一系列指令 在C#中,迭代器是一种特殊的结构,允许…

炸砖块游戏的最终图案

描述 小红正在玩一个“炸砖块”游戏,游戏的规则如下:初始有一个 n * m 的砖块矩阵。小红会炸 k 次,每次会向一个位置投炸弹,如果这个位置有一个砖块,则砖块消失,上方的砖块向下落。小红希望你画出最终砖块的图案。 输入描述 第一行输入三个正整数 n, m, k,代表矩阵的行…

【Python项目】图像信息隐藏技术的实现

【Python项目】图像信息隐藏技术的实现 技术简介:采用Python技术、MYSQL数据库等实现。 系统简介:系统主要的功能有登录模块、对个人信息的维护模块、图片信息的隐藏模块涉及图片的上传和信息的验证。 背景: 在当今时代,信息技术…

游戏市场成果及趋势

2024 年的游戏行业发展情况如何?这是一个既关系到开发商,又关系到玩家的问题,而市场分析师可以为我们揭晓答案。下面,就让我们来看看分析师给出的结论以及他们对未来趋势的预测。 玩家 自 2021 年起,全球平均游戏时间…

Cursor使用教程

一、安装 二、使用 1、操作对象 项目 文件 文件块 代码块 终端 2、chat 窗口 提问 ctrl回车 会针对本地代码进行提问 1、快捷键 ctrl i 打开对话,可以指定文本询问 ctrl k 在文本中询问 ctrl y 接受生成 ctr l n …

大数据原生集群 (Hadoop3.X为核心) 本地测试环境搭建二

本篇安装软件版本 mysql5.6 spark3.2.1-hadoop3.2 presto0.272 zeppelin0.11.2 kafka_2.13_3.7.2 mysql 安装步骤见-》 https://blog.csdn.net/dudadudadd/article/details/110874570 spark 安装步骤见-》https://blog.csdn.net/dudadudadd/article/details/109719624 安装…

【算法学习】——整数划分问题详解(动态规划)

🧮整数划分问题是一个较为常见的算法题,很多问题从整数划分这里出发,进行包装,形成新的题目,所以完全理解整数划分的解决思路对于之后的进一步学习算法是很有帮助的。 「整数划分」通常使用「动态规划」解决&#xff0…

深入理解计算机系统阅读笔记-第十二章

第12章 网络编程 12.1 客户端-服务器编程模型 每个网络应用都是基于客户端-服务器模型的。根据这个模型,一个应用时由一个服务器进程和一个或者多个客户端进程组成。服务器管理某种资源,并且通过操作这种资源来为它的客户端提供某种服务。例如&#xf…

MySQL | GROUP BY子句使用详解

关注:CodingTechWork 引言 在 MySQL 中,GROUP BY 子句用于将查询结果按照一个或多个列进行分组。每个分组会返回一行,通常与聚合函数(如 COUNT(), SUM(), AVG() 等)一起使用,用于汇总每个分组的数据。  …

Redis 实战篇 ——《黑马点评》(中)

《引言》 (中)篇将接着记录 Redis 实战篇 ——《黑马点评》(上)篇之后的学习内容与笔记,希望大家能够点赞、收藏支持一下 ᕦ(・ㅂ・)ᕤ,谢谢大家。 传送门(上)&…