未提供项目说明
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
,以存储输入文件路径。但是,你
也可以使用state
dict属性来存储任意状态。对于需要一些设置和拆卸步骤的任务,
使用Task._preamble
和Task._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中,^{__or__
的两个对象上使用按位or运算符时调用。在
task-py
利用这种灵活性,允许管道Task
一起创建管道。例如,假设
您正在编写一个CSV处理工具包,并希望创建一个命令行应用程序,从CSV读取数据并删除
一些专栏。这条管道有两个明确的步骤:
- 将CSV中的数据读入某些数据结构
- 删除指定的列(并写入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()
贡献/建议
欢迎投稿和建议!要提出功能请求,请报告错误,或对现有的 功能,请提交问题。如需投稿,请提交一份个人简历,但请务必进行打印、类型检查和测试 你的代码在做之前所以。提前谢谢!在
- 项目
标签: