用 Python 創(chuàng)建你自己的 Shell

subprocess.run對其自身進(jìn)行延遲求值并保存結(jié)果。import?subprocess
class?PipePy:
????def?__init__(self,?*args):
????????self._args?=?args
????????self._result?=?None
????def?_evaluate(self):
????????if?self._result?is?not?None:
????????????return
????????self._result?=?subprocess.run(self._args,
??????????????????????????????????????capture_output=True,
??????????????????????????????????????text=True)
????@property
????def?returncode(self):
????????self._evaluate()
????????return?self._result.returncode
????@property
????def?stdout(self):
????????self._evaluate()
????????return?self._result.stdout
????def?__str__(self):
????????return?self.stdout
????@property
????def?stderr(self):
????????self._evaluate()
????????return?self._result.stderr
ls?=?PipePy('ls')
ls_l?=?PipePy('ls',?'-l')
print(ls)
#?<<
#?...?main.py
#?...?tags
print(ls_l)
#?<<
#?...?-rw-r--r--?1?kbairak?kbairak??125?Jan?22?08:53?files.txt
#?...?-rw-r--r--?1?kbairak?kbairak?5425?Feb??1?21:54?main.py
#?...?-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
ls_l?=?PipePy('ls',?'-l')
print(ls_l)
ls?=?PipePy('ls')
print(ls('-l'))
PipePy('ls',?'-l')
PipePy('ls')('-l')
class?PipePy:
????#?__init__,?etc
????def?__call__(self,?*args):
????????args?=?self._args?+?args
????????return?self.__class__(*args)
ls?=?PipePy('ls')
print(ls('-l'))
#?<<
#?...?-rw-r--r--?1?kbairak?kbairak??125?Jan?22?08:53?files.txt
#?...?-rw-r--r--?1?kbairak?kbairak?5425?Feb??1?21:54?main.py
#?...?-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
ls傳遞更多參數(shù),則可能會(huì)遇到--sort = size。我們可以輕松地執(zhí)行ls('-l','--sort = size')。我們可以做得更好嗎??class?PipePy:
-????def?__init__(self,?*args):
+????def?__init__(self,?*args,?**kwargs):
?????????self._args?=?args
+????????self._kwargs?=?kwargs
?????????self._result?=?None
?????def?_evaluate(self):
?????????if?self._result?is?not?None:
?????????????return
-????????self._result?=?subprocess.run(self._args,
+????????self._result?=?subprocess.run(self._convert_args(),
???????????????????????????????????????capture_output=True,
???????????????????????????????????????text=True)
?
+????def?_convert_args(self):
+????????args?=?[str(arg)?for?arg?in?self._args]
+????????for?key,?value?in?self._kwargs.items():
+????????????key?=?key.replace('_',?'-')
+????????????args.append(f"--{key}={value}")
+????????return?args
?
-????def?__call__(self,?*args):
+????def?__call__(self,?*args,?**kwargs):
?????????args?=?self._args?+?args
+????????kwargs?=?{**self._kwargs,?**kwargs}
-????????return?self.__class__(*args)
+????????return?self.__class__(*args,?**kwargs)
?????#?returncode,?etc
print(ls('-l'))
#?<<
#?...?-rw-r--r--?1?kbairak?kbairak??125?Jan?22?08:53?files.txt
#?...?-rw-r--r--?1?kbairak?kbairak?5425?Feb??1?21:54?main.py
#?...?-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
print(ls('-l',?sort="size"))
#?<<
#?...?-rw-r--r--?1?kbairak?kbairak?5425?Feb??1?21:54?main.py
#?...?-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
#?...?-rw-r--r--?1?kbairak?kbairak??125?Jan?22?08:53?files.txt
ls?=?PipePy('ls')
grep?=?PipePy('grep')
print(ls?|?grep('tags'))
#?<<
__init__和__call__方法接受一個(gè)僅用于關(guān)鍵字的新_pipe_input關(guān)鍵字參數(shù),該參數(shù)將保存在self上。_pipe_input,它將作為輸入?yún)?shù)傳遞給subprocess.run。__or__方法以將左操作數(shù)的結(jié)果作為pipe輸入傳遞給右操作數(shù)。?class?PipePy:
-????def?__init__(self,?*args,?**kwargs):
+????def?__init__(self,?*args,?_pipe_input=None,?**kwargs):
?????????self._args?=?args
?????????self._kwargs?=?kwargs
+????????self._pipe_input?=?_pipe_input
?????????self._result?=?None
?
-????def?__call__(self,?*args,?**kwargs):
+????def?__call__(self,?*args,?_pipe_input=None,?**kwargs):
?????????args?=?self._args?+?args
?????????kwargs?=?{**self._kwargs,?**kwargs}
-????????return?self.__class__(*args,?**kwargs)
+????????return?self.__class__(*args,?_pipe_input=_pipe_input,?**kwargs)
?
?????def?_evaluate(self):
?????????if?self._result?is?not?None:
?????????????return
?????????self._result?=?subprocess.run(self._convert_args(),
+??????????????????????????????????????input=self._pipe_input,
???????????????????????????????????????capture_output=True,
???????????????????????????????????????text=True)
?
+????def?__or__(left,?right):
+????????return?right(_pipe_input=left.stdout)
ls?=?PipePy('ls')
grep?=?PipePy('grep')
print(ls('-l')?|?grep('tags'))
#?<<-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
class?PipePy:
????#?__init__,?etc
????def?__bool__(self):
????????return?self.returncode?==?0
git?=?PipePy('git')
grep?=?PipePy('grep')
if?git('branch')?|?grep('my_feature'):
????print("Branch?'my_feature'?found")
class?PipePy:
????#?__init__,?etc
????def?__gt__(self,?filename):
????????with?open(filename,?'w')?as?f:
????????????f.write(self.stdout)
????def?__rshift__(self,?filename):
????????with?open(filename,?'a')?as?f:
????????????f.write(self.stdout)
????def?__lt__(self,?filename):
????????with?open(filename)?as?f:
????????????return?self(_pipe_input=f.read())
ls?=?PipePy('ls')
grep?=?PipePy('grep')
cat?=?PipePy('cat')
ls?>?'files.txt'
print(grep('main')?'files.txt')
#?<<
ls?>>?'files.txt'
print(cat('files.txt'))
#?<<
#?...?main.py
#?...?tags
#?...?files.txt
#?...?main.py
#?...?tags
class?PipePy:
????#?__init__,?etc
????def?__iter__(self):
????????return?iter(self.stdout.split())
ls?=?PipePy('ls')
for?name?in?ls:
????print(name.upper())
#?<<
#?...?MAIN.PY
#?...?TAGS
class?PipePy:
????#?__init__,?etc
????def?as_table(self):
????????lines?=?self.stdout.splitlines()
????????fields?=?lines[0].split()
????????result?=?[]
????????for?line?in?lines[1:]:
????????????item?=?{}
????????????for?i,?value?in?enumerate(line.split(maxsplit=len(fields)?-?1)):
????????????????item[fields[i]]?=?value
????????????result.append(item)
????????return?result
ps?=?PipePy('ps')
print(ps)
#?<<????PID?TTY??????????TIME?CMD
#?...????4205?pts/4????00:00:00?zsh
#?...???13592?pts/4????00:00:22?ptipython
#?...???16253?pts/4????00:00:00?ps
ps.as_table()
#?<<[{'PID':?'4205',?'TTY':?'pts/4',?'TIME':?'00:00:00',?'CMD':?'zsh'},
#?...??{'PID':?'13592',?'TTY':?'pts/4',?'TIME':?'00:00:22',?'CMD':?'ptipython'},
#?...??{'PID':?'16208',?'TTY':?'pts/4',?'TIME':?'00:00:00',?'CMD':?'ps'}]
bash實(shí)用程序:import?os
cd?=?os.chdir
export?=?os.environ.__setitem__
pwd?=?PipePy('pwd')
pwd
#?<</home/kbairak/prog/python/pipepy
cd('..')
pwd
#?<</home/kbairak/prog/python
ls并完成它。class?PipePy:
????#?__init__,?etc
????def?__repr__(self):
????????return?self.stdout?+?self.stderr
shell>>>?ls?=?PipePy('ls')
>>>?ls
files.txt
main.py
tags
from?pipepy?import?PipePy
tar?=?PipePy('tar')
tar('-xf',?'some_archive.tar')
print("File?extracted")
tar調(diào)用實(shí)際上并未得到評估。我認(rèn)為一個(gè)不錯(cuò)的慣例是,如果不帶參數(shù)調(diào)用__call__強(qiáng)制求值:?class?PipePy:
?????def?__call__(self,?*args,?_pipe_input=None,?**kwargs):
?????????args?=?self._args?+?args
?????????kwargs?=?{**self._kwargs,?**kwargs}
-????????return?self.__class__(*args,?_pipe_input=_pipe_input,?**kwargs)
+????????result?=?self.__class__(*args,?_pipe_input=_pipe_input,?**kwargs)
+????????if?not?args?and?not?_pipe_input?and?not?kwargs:
+????????????result._evaluate()
+????????return?result
?from?pipepy?import?PipePy
?tar?=?PipePy('tar')
-tar('-xf',?'some_archive.tar')
+tar('-xf',?'some_archive.tar')()
?print("File?extracted")
date?=?PipePy('date')
date
#?<<
#?Wait?5?seconds
date
#?<<
date沒有改變。date對象將其_result保留在內(nèi)存中。隨后的評估實(shí)際上不會(huì)調(diào)用該命令,而只是返回存儲(chǔ)的值。date?=?PipePy('date')
date()
#?<<
#?Wait?5?seconds
date()
#?<<
PipePy構(gòu)造函數(shù)返回的實(shí)例不應(yīng)該是惰性的,但由__call__調(diào)用返回的實(shí)例將是惰性的。?class?PipePy:
-????def?__init__(self,?*args,?_pipe_input=None,?**kwargs):
+????def?__init__(self,?*args,?_pipe_input=None,?_lazy=False,?**kwargs):
?????????self._args?=?args
?????????self._kwargs?=?kwargs
?????????self._pipe_input?=?_pipe_input
+????????self._lazy?=?_lazy
?????????self._result?=?None
?
?????def?__call__(self,?*args,?_pipe_input=None,?**kwargs):
?????????args?=?self._args?+?args
?????????kwargs?=?{**self._kwargs,?**kwargs}
-????????result?=?self.__class__(*args,?_pipe_input=_pipe_input,?**kwargs)
+????????result?=?self.__class__(*args,
+????????????????????????????????_pipe_input=_pipe_input,
+????????????????????????????????_lazy=True,
+????????????????????????????????**kwargs)
?????????if?not?args?and?not?_pipe_input?and?not?kwargs:
?????????????result._evaluate()
?????????return?result
?
?????def?_evaluate(self):
-????????if?self._result?is?not?None:
+????????if?self._result?is?not?None?and?self._lazy:
?????????????return
?????????self._result?=?subprocess.run(self._convert_args(),
???????????????????????????????????????input=self._pipe_input,
???????????????????????????????????????capture_output=True,
???????????????????????????????????????text=True)
date?=?PipePy('date')
date
#?<<
#?Wait?5?seconds
date
#?<<
date?=?PipePy('date')
d?=?date()
d
#?<<
#?Wait?5?seconds
d
#?<<
d會(huì)更新其值。ls('-l')不錯(cuò),但是如果我們像人類一樣簡單地做ls -l,那就太好了。嗯,我有個(gè)主意:class?PipePy:
????#?__init__,?etc
????def?__sub__(left,?right):
????????return?left(f"-{right}")
ls?=?PipePy('ls')
ls?-?'l'
#?<<
#?...?-rw-r--r--?1?kbairak?kbairak???46?Feb??1?23:04?files.txt
#?...?-rw-r--r--?1?kbairak?kbairak?5425?Feb??1?21:54?main.py
#?...?-rw-r--r--?1?kbairak?kbairak?1838?Feb??1?21:54?tags
l?=?'l'
ls?-l
import?string
for?char?in?string.ascii_letters:
????if?char?in?locals():
????????continue
????locals()[char]?=?char
class?PipePy:
????#?__init__,?etc
locals()給了我一個(gè)靈感。為什么我們必須一直實(shí)例化PipePy?我們無法在路徑中找到所有可執(zhí)行文件,并根據(jù)它們創(chuàng)建PipePy實(shí)例嗎?我們當(dāng)然可以!import?os
import?stat
for?path?in?os.get_exec_path():
????try:
????????names?=?os.listdir(path)
????except?FileNotFoundError:
????????continue
????for?name?in?names:
????????if?name?in?locals():
????????????continue
????????if?'x'?in?stat.filemode(os.lstat(os.path.join(path,?name)).st_mode):
????????????locals()[name]?=?PipePy(name)
from?pipepy?import?mysqladmin,?sleep,?drush,?grep
for?i?in?range(10):
????if?mysqladmin('ping',
??????????????????host="mysql_drupal7",
??????????????????user="user",
??????????????????password="password"):
????????break
????sleep(1)()??#?Remember?to?actually?invoke
if?not?drush('status',?'bootstrap')?|?grep('-q',?'Successful'):
????drush('-y',?'site-install',?'standard',
??????????db_url="mysql://user:password@mysql_drupal7:3306/drupal",
??????????acount_pass="kbairak")()??#?Remember?to?actually?invoke
drush('en',?'tmgmt_ui',?'tmgmt_entity_ui',?'tmgmt_node_ui')()
更多閱讀
特別推薦

點(diǎn)擊下方閱讀原文加入社區(qū)會(huì)員
評論
圖片
表情
