Python 全栈系列241 GFGo Lite迭代

2024-05-06 15:28
文章标签 python 系列 全栈 迭代 lite 241 gfgo

本文主要是介绍Python 全栈系列241 GFGo Lite迭代，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

说明

随着整个算网开发逐渐深入，各个组件、微服务的数量、深度在不断增加。由于算网是个人项目，我一直按照MVP(Minimum Viable Product )的原则在推进。由于最初的时候对架构、算法和业务的理解并没有那么深刻，所以MVP的内容还是在不断变化（增加）的。比较幸运的是，中间走的弯路比较少，整体方向上一直没有大偏差，应该可以在预期的时间内达到目标。

从工具和使用的角度来看，我一边在算网的概念下构造工具，一边在尝试使用这些工具提高我的效率/能力上限。过去，由于已知的、明确要实现的功能就很多，所以一直是不断急速前进，常见的一种情况是：工具开发好了，测试也好用，但是就扔在一边，继续开发别的新工具去了。

现在的情况是：完成一个MVP所需要新开发的功能几乎没有了，反而是将已有的工具用起来比较重要。

今年做的比较重要的改变是使用streamlit、gradio将微服务前后端一体化。这种方式最初是大语言模型广泛采用的，的确非常方便。对算网而言，大量的已开发组件和功能，将通过这种方式进行文档、测试和使用的一体化。

在这里插入图片描述

总之，现在的重点是实现(Realization)。

内容

1 GlobalFunc更新

1.1 程序

目前仍然采用vscode代码开发的方式，创建一个个新的py文件。

1.2 推送更新

步骤如下：

  • 1 切到项目路径下cd .../m4git/GlobalFunc
  • 2 导入操作对象
from op044_GlobalFunc_01BaseOpt import gfbase# 1 查看当前分枝
gfbase._get_current_branch()
# 2 扫描所有文件的信息
scan_dict = gfbase._get_scan_files()
# 3 提交git项目
gfbase._simple_commit_git()
# 4 刷新一个包的初始化文件
gfbase._generate_init_py('Base')
gfbase._generate_init_py('Parse')
gfbase._generate_init_py('TFIDF')
# 5 批量存储函数
some_pack_list = [x for x in scan_dict.keys() if x.startswith('Base.') or x.startswith('Parse.') or x.startswith('TFIDF.')]for some_file in some_pack_list:gfbase._save_a_file_rom(some_file)

在这里插入图片描述

2 GFGo Lite 更新

GFGo Lite 有所修改，文件在项目文件夹gfgo_lite_24090 下面

2.1 文件

切入gfgo_lite_build容器，里面已经有了一些依赖文件

  • 1 init_funcs 里面存放了数据库连接相关的函数
  • 2 按模块执行加载，然后初始化

以下是从Base模块中加载一些函数的示例

from init_funcs import create_file,generate_init_py,RedisOrMongo, Naiveimport os 
import json
# 声明空间# 在容器中启动
redis_cfg = Naive()
redis_cfg.redis_agent_host = 'http://172.17.0.1:24021/'
redis_cfg.redis_connection_hash = None# 模块Base
if True:# 声明当前要创建的程序文件夹：默认为funcstarget_folder = 'GlobalFunc'tier1 = 'sp_GlobalFunc'var_space_name = 'Base'# 分支，一般默认为masterbranch_name = 'master' tier2 = '_'.join([var_space_name, branch_name])the_space_name = '.'.join([tier1,tier2])target_folder = target_folder + '/' + var_space_nameos.makedirs(target_folder, exist_ok=True)rom = RedisOrMongo(the_space_name, redis_cfg.dict(),backend='mongo', mongo_servername='m7.24065')# 这个一般根据需要，或者代码中得来 --- 需要的列表项func_list = [ 'from_pickle','to_pickle','is_file_exists','gen_time_axis','ATimer2','get_time_str1','cols2s','create_folder_if_notexist','flat_dict','flatten_list']# func_list = [ 'from_pickle','to_pickle','pose_a_file']for some_name in func_list:# 获取 meta,data : data就是代码字符the_data = rom.getx(some_name)filename = the_data['meta']['name']filedata = the_data['data']create_file(target_folder, filename, filedata)# 生成初始化文件generate_init_py(target_folder)

在导入新的包时，需要手动修改GlobalFunc下的__init__.py(与Base包和Parse包平级)。

from . import Base
from . import Parse

2.2 服务

在 /workspace下直接启动服务即可 python3 server.py。

3 应用

3.1 通用函数 UCS

UCS是一个规范，为了支持这个规范，必须依赖一些特定(同时也是固定)的函数。因为函数的通用性，所以这些函数放在了最外层，每个函数都占据了一个api接口。

3.1.1 block (int)

整型block分割三件套

import requests as req some_dict = {}
some_dict['rec_id'] = 111111res = req.post('http://127.0.0.1:8000/get_brick_name/', json = some_dict).json()'0.0.0.11'some_dict = {}
some_dict['brick_name'] = '0.0.0.11'res = req.post('http://127.0.0.1:8000/get_brick_bounds/', json = some_dict).json()
[110000.0, 120000.0]some_dict = {}
some_dict['start_brick_name'] = '0.0.0.9'
some_dict['end_brick_name'] = '0.0.0.12'res = req.post('http://127.0.0.1:8000/get_brick_list/', json = some_dict).json()
['0.0.0.9', '0.0.0.10', '0.0.0.11'
3.1.2 time block

操作如下：时间支持字符和数值(时间戳)两种模式。

import requests as req # char
some_dict = {}
some_dict['dt_str_or_ts'] = '2024-01-31 11:11:11'
res = req.post('http://127.0.0.1:8000/get_time_brick_name/', json = some_dict).json()
'2024.01.31.11'# num
some_dict = {}
some_dict['dt_str_or_ts'] = 1706670671
res = req.post('http://127.0.0.1:8000/get_time_brick_name/', json = some_dict).json()
'2024.01.31.11'# char 
some_dict = {}
some_dict['brick_name'] = '2024.01.31.11'
some_dict['char_or_num'] = 'char'
res = req.post('http://127.0.0.1:8000/get_time_brick_bounds/', json = some_dict).json()'''
In [13]: res
Out[13]: ['2024-01-31 11:00:00', '2024-01-31 12:00:00']
'''# num 
some_dict = {}
some_dict['brick_name'] = '2024.01.31.11'
some_dict['char_or_num'] = 'num'
res = req.post('http://127.0.0.1:8000/get_time_brick_bounds/', json = some_dict).json()'''
In [15]: res
Out[15]: [1706670000, 1706673600]
'''some_dict = {}
some_dict['start_brick_name'] = '2024.01.31.11'
some_dict['end_brick_name'] = '2024.02.02.11'
res = req.post('http://127.0.0.1:8000/get_time_brick_list/', json = some_dict).json()'''
In [11]: res
Out[11]:
['2024.01.31.11','2024.01.31.12','2024.01.31.13','2024.01.31.14',...
'''

3.2 功能函数

3.2.1 Base包的函数调用

以下是两个Base包函数的测试

import requests as req # 测试1：调用Base包的函数
kwargs = {'ts':None, 'bias_hours':8}
pack_func = 'Base.get_time_str1'some_dict = {}
some_dict['kwargs'] = kwargs
some_dict['pack_func'] = pack_funcres = req.post('http://127.0.0.1:8000/gfgo/', json = some_dict).json()
'2024-05-05 11:04:35'# 测试2：列表扁平化
kwargs = {'nested_list':[[1,2],[3],[4,5]]}
pack_func = 'Base.flatten_list'some_dict = {}
some_dict['kwargs'] = kwargs
some_dict['pack_func'] = pack_funcres = req.post('http://127.0.0.1:8000/gfgo/', json = some_dict).json()
[1, 2, 3, 4, 5]

有两点需要注意：

  • 1 函数规范为全部关键字参数输入（主要是为了方便调用）
  • 2 接口直接返回处理信息（而不是包上状态和消息）
3.3.3 Parse包的函数调用
x = "This is a sample text."
word_list = ["sample", "test", "string"]kwargs = {'x':x, 'word_list':word_list}
pack_func = 'Parse.judge_existence'some_dict = {}
some_dict['kwargs'] = kwargs
some_dict['pack_func'] = pack_funcres = req.post('http://127.0.0.1:8000/gfgo/', json = some_dict).json()True

3.4 服务迭代

推送新的变化
docker push myregistry.domain.com:24052/worker.andy.gfgo_lite_24090:v101
启动服务

docker run -d \--restart=always \--name=gfgo_lite_24090 \-v /etc/localtime:/etc/localtime  \-v /etc/timezone:/etc/timezone\-v /etc/hostname:/etc/hostname\-e "LANG=C.UTF-8" \-w /workspace\-p 24090:8000\myregistry.domain.com:24052/worker.andy.gfgo_lite_24090:v101 \sh -c "python3 server.py"

公网调用

In [3]:...: some_dict = {}...: some_dict['brick_name'] = '2024.01.31.11'...: some_dict['char_or_num'] = 'num'...: res = req.post('http://WAN_IP:24090/get_time_brick_bounds/', json = some_dict).json()In [4]: res
Out[4]: [1706670000, 1706673600]

4 总结与展望

RuleSet As A Func

将复杂的规则（判定）作为一个函数调用。

Series Apply

每个函数都要支持列表（多个元素）的并行处理。

踩过的一个小坑：GlobalFunc使用了一个公网机的Redis做ROM，而GFGoLite使用m7本地的redis，导致了逻辑上看起来更新了，但是实际未更新。

这篇关于Python 全栈系列241 GFGo Lite迭代的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/964710。

相关文章

Python调用Orator ORM进行数据库操作

Python调用Orator ORM进行数据库操作

《Python调用OratorORM进行数据库操作》OratorORM是一个功能丰富且灵活的PythonORM库,旨在简化数据库操作,它支持多种数据库并提供了简洁且直观的API,下面我们就... 目录Orator ORM 主要特点安装使用示例总结Orator ORM 是一个功能丰富且灵活的 python O
阅读更多...
Python使用国内镜像加速pip安装的方法讲解

Python使用国内镜像加速pip安装的方法讲解

《Python使用国内镜像加速pip安装的方法讲解》在Python开发中,pip是一个非常重要的工具,用于安装和管理Python的第三方库,然而,在国内使用pip安装依赖时,往往会因为网络问题而导致速... 目录一、pip 工具简介1. 什么是 pip？2. 什么是 -i 参数？二、国内镜像源的选择三、如何
阅读更多...
python使用fastapi实现多语言国际化的操作指南

python使用fastapi实现多语言国际化的操作指南

《python使用fastapi实现多语言国际化的操作指南》本文介绍了使用Python和FastAPI实现多语言国际化的操作指南,包括多语言架构技术栈、翻译管理、前端本地化、语言切换机制以及常见陷阱和... 目录多语言国际化实现指南项目多语言架构技术栈目录结构翻译工作流1. 翻译数据存储2. 翻译生成脚本
阅读更多...
如何通过Python实现一个消息队列

如何通过Python实现一个消息队列

《如何通过Python实现一个消息队列》这篇文章主要为大家详细介绍了如何通过Python实现一个简单的消息队列,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录如何通过 python 实现消息队列如何把 http 请求放在队列中执行1. 使用 queue.Queue 和 reque
阅读更多...
Python如何实现PDF隐私信息检测

Python如何实现PDF隐私信息检测

《Python如何实现PDF隐私信息检测》随着越来越多的个人信息以电子形式存储和传输,确保这些信息的安全至关重要,本文将介绍如何使用Python检测PDF文件中的隐私信息,需要的可以参考下... 目录项目背景技术栈代码解析功能说明运行结php果在当今，数据隐私保护变得尤为重要。随着越来越多的个人信息以电子形
阅读更多...
使用Python快速实现链接转word文档

使用Python快速实现链接转word文档

《使用Python快速实现链接转word文档》这篇文章主要为大家详细介绍了如何使用Python快速实现链接转word文档功能,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 演示代码展示from newspaper import Articlefrom docx import
阅读更多...
Python Jupyter Notebook导包报错问题及解决

Python Jupyter Notebook导包报错问题及解决

《PythonJupyterNotebook导包报错问题及解决》在conda环境中安装包后,JupyterNotebook导入时出现ImportError,可能是由于包版本不对应或版本太高,解决方... 目录问题解决方法重新安装Jupyter NoteBook 更改Kernel总结问题在conda上安装了
阅读更多...
Python如何计算两个不同类型列表的相似度

Python如何计算两个不同类型列表的相似度

《Python如何计算两个不同类型列表的相似度》在编程中,经常需要比较两个列表的相似度,尤其是当这两个列表包含不同类型的元素时,下面小编就来讲讲如何使用Python计算两个不同类型列表的相似度吧... 目录摘要引言数字类型相似度欧几里得距离曼哈顿距离字符串类型相似度Levenshtein距离Jaccard相
阅读更多...
Python安装时常见报错以及解决方案

Python安装时常见报错以及解决方案

《Python安装时常见报错以及解决方案》：本文主要介绍在安装Python、配置环境变量、使用pip以及运行Python脚本时常见的错误及其解决方案,文中介绍的非常详细,需要的朋友可以参考下... 目录一、安装 python 时常见报错及解决方案（一）安装包下载失败（二）权限不足二、配置环境变量时常见报错及
阅读更多...
Python中顺序结构和循环结构示例代码

Python中顺序结构和循环结构示例代码

《Python中顺序结构和循环结构示例代码》：本文主要介绍Python中的条件语句和循环语句,条件语句用于根据条件执行不同的代码块,循环语句用于重复执行一段代码,文章还详细说明了range函数的使... 目录一、条件语句（1）条件语句的定义（2）条件语句的语法（a）单分支 if（b）双分支 if-else（
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2