未提供项目说明

lc-task的Python项目详细描述


task-py

概述

有一天你需要写一个脚本来处理一些数据,所以你坐下来写。文件路径和其他参数 都是硬编码的,但很管用。几天后,您需要执行相同的操作,但需要接受 命令行,可能还有一个参数,所以您可以将一些功能重构为函数并创建一个简单的CLI 使用sys.argv。然后一个同事问他/她/他们是否可以使用这个脚本,所以您最终分解并使用argparse 编写更好的CLI。如果这个循环听起来很熟悉,task-py可以帮助您跳过第一步和第二步,而 编写可互操作、可组合的工具。在

安装

从PyPi安装(首选方法)

pip install lc-task

使用Pip

从GitHub安装 ^{pr2}$

其中x.x.x是要下载的版本。在

手动下载安装

要下载源分发和/或控制盘文件,请导航到 https://github.com/libcommon/task-py/tree/releases/vx.x.x/dist,其中x.x.x是要安装的版本, 并通过用户界面或wget等工具进行下载。然后安装运行:

pip install <downloaded file>

下载后请不要更改文件名,因为Pip需要安装文件的特定命名约定。在

依赖关系

task-py没有外部依赖项。官方只支持Python>;=3.6版本。在

入门

创建任务

创建Task很简单:指定一些(可选)属性并实现Task._perform_task,它执行主 工作单位。例如,您可以实现一个采用文件路径并打印该路径的数据的任务 文件到标准输出:

frompathlibimportPathfromlc_taskimportTaskclassCatFileTask(Task):"""Print contents of a file to stdout."""__slots__=("_input_path",)def__init__(self,input_path:Path,*args,**kwargs)->None:super().__init__(*args,**kwargs)self._input_pathdef_perform_task(self)->None:ifnotself._input_path.is_file():raiseFileNotFoundError(str(self._input_path))withself._input_path.open()asinput_file:print(input_file.read())

在上面的例子中,我们在CatFileTask上定义了一个属性_input_path,以存储输入文件路径。但是,你 也可以使用statedict属性来存储任意状态。对于需要一些设置和拆卸步骤的任务, 使用Task._preambleTask._postamble函数,它们在主要工作单元完成之前和之后被调用。 要显式地运行此任务,您将使用Task.run方法。在

接受命令行输入

要创建接受命令行输入的单个任务(使用argparse),请创建一个CLITask的子类来定义 命令的名称、简要说明和命令行参数。以上面的例子为例:

frompathlibimportPathfromlc_taskimportCLITaskclassCatFileTask(CLITask):__slots__=("_input_path",)COMMAND="cat"DESCRIPTION="Print contents of a file to stdout."@classmethoddefgen_command_parser(cls,parser:Optional[ArgumentParser]=None)->ArgumentParser:parser=super().gen_command_parser()parser.add_argument("_input_path",type=Path,metavar="INPUT_PATH",help="Path to input file")returnparserdef_perform_task(self)->None:ifnotself._input_path.is_file():raiseFileNotFoundError(str(self._input_path))withself._input_path.open()asinput_file:print(input_file.read())if__name__=="__main__":CatFileTask.run_command()

要用子命令定义更复杂的层次化命令行接口,请看一下docstring 对于cli.gen_cli_parser。它允许你用命令到动作的映射来定义你的命令行应用程序, 然后为您生成CLI。在

管道和消息传递

任何使用过Bash、PowerShell或其他脚本语言的人都熟悉可组合性和管道的概念: 编写简单的命令来返回由其他命令识别的结构化数据可能非常强大。Python没有 支持这种开箱即用的编程风格,但它确实支持为自定义类型重载运算符,例如 按位或运算符(|)。在Python中,^{} builtin method 在第一个(左侧)没有实现__or__的两个对象上使用按位or运算符时调用。在

task-py利用这种灵活性,允许管道Task一起创建管道。例如,假设 您正在编写一个CSV处理工具包,并希望创建一个命令行应用程序,从CSV读取数据并删除 一些专栏。这条管道有两个明确的步骤:

  1. 将CSV中的数据读入某些数据结构
  2. 删除指定的列(并写入stdout)
importcsvfrompathlibimportPathfromtypingimportList,Optional,TextIOfromlc_taskimportCLITask,Task,TaskResultclassCSVColumnRemovalTask(Task):"""Remove specified columns from frows in a CSV."""__slots__=("columns","input_file","reader")def__init__(self,*args,columns:Optional[List[str]]=None,**kwargs)->None:super().__init__(*args,**kwargs)self.columns=columnsdef_perform_task(self)->None:forrecordinreader:forcolumninself.columns:delrecord[column]print(", ".join(record))def_preamble(self)->None:self.input_file.close()classCSVReaderTaskResult(TaskResult):__slots__=("input_file","reader")def__init__(self:input_file:Optional[TextIO]=None,reader:Optional[csv.DictReader]=None,**kwargs)->None:super().__init__(**kwargs)self.input_file=input_fileself.reader=readerclassCSVReaderTask(Task):"""Read data from a CSV into a `csv.DictReader` object."""__slots__=("input_path",)@staticmethoddefgen_task_result()->TaskResult:returnCSVReaderTaskResult()def__init__(self,*args,input_path:Optional[Path]=None,**kwargs)->None:super().__init__(*args,**kwargs)self.input_path=input_pathdef_perform_task(self)->None:# Need to check self.input_path because it could be provided in# __init__ or via merge_objectifnotself.input_path:raiseValueError("must provide path to CSV file")input_file=self.input_path.open(newline="")reader=csv.DictReader(input_file,dialect=csv.unix_dialect)# Using a defined TaskResult type here, but could also just use a dictself.result=CSVReaderTaskResult(input_file=input_file,reader=reader)classRemoveColumnsCLITask(CLITask):__slots__=("columns","input_path")COMMAND="remove-columns"DESCRIPTION="Remove columns from a CSV and write to disk."@classmethoddefgen_command_parser(cls,parser:Optional[ArgumentParser]=None)->ArgumentParser:parser=super().gen_command_parser()parser.add_argument("input_path",type=Path,help="Path to input file")parser.add_argument("columns",nargs=+,help="Column names to remove")returnparserdef_perform_task(self)->None:# Pipe the output of CSVReaderTask to CSVColumnRemovalTask(CSVReaderTask(input_path=self.input_path).run()|CSVColumnRemovalTask(columns=self.columns))if__name__=="__main__":RemoveColumnsCLITask.run_command()

贡献/建议

欢迎投稿和建议!要提出功能请求,请报告错误,或对现有的 功能,请提交问题。如需投稿,请提交一份个人简历,但请务必进行打印、类型检查和测试 你的代码在做之前所以。提前谢谢!在

欢迎加入QQ群-->: 979659372 Python中文网_新手群

推荐PyPI第三方库


热门话题
java得到的分数是Android屏幕大小的25%   爪哇杰克逊。数据绑定。JsonNode()在null字段上返回true(不应该返回)   java使用改型2获取和获取未知数据集   java Play框架:使用WS将图像发布到imageshack   高效eclipse的java最佳设置   swing KeyListener Java   无法解析java链表   java ListView内存问题   java在创建重载构造函数、实例化对象时遇到问题   如何在java中沿树进行预排序遍历,并打印0和1以对应每个节点上的特定字符?   swing在Java中使用矩阵和JButton设计扑克UI   SpringREST异常无法读取JSON:无法反序列化java实例。lang.从START_对象中取出的字符串   java如何消除JPanel的透明度   为什么“i”和“i”在Java for循环中有相同的行为?   java静态工厂方法   java在不使用Stringbuilder的情况下将SQLite数据设置为JsonObject类   java能否为标题“DispositionNotificationTo”分配2个或更多地址?   java无法打开openclinica