python 运维包_基础入门_Python-模块和包.运维开发中__import__动态导入最佳实践?
常規(guī)導(dǎo)入:import?module_name[,module1,...]
from?module_name?import?[*|child[,child1,...]
from?module_name?import?[*|child[,child1,...]?as?alias_name
注意: 導(dǎo)入語句可出現(xiàn)在程序任意位置,自定義包要實現(xiàn)from module_name import *的效果則此模塊必須在__init__.py實現(xiàn)__all__ = ['module_1', 'module_2']
加載一次:
說明: 多次重復(fù)使用import語句時,不會重新加載模塊,而是把該模塊的內(nèi)存地址給引用到本地環(huán)境變量==>?x.py?<==
#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://xmdevops.blog.51cto.com/
#?Purpose:
#
"""
#?說明:?導(dǎo)入公共模塊
import?os
#?說明:?導(dǎo)入其它模塊
print?'os?in?x.py',?id(os)
==>?y.py?<==
#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://xmdevops.blog.51cto.com/
#?Purpose:
#
"""
#?說明:?導(dǎo)入公共模塊
import?x
import?os
#?說明:?導(dǎo)入其它模塊
if?__name__?==?'__main__':
print?'os?in?y.py',?id(os)
import?x
重新加載:
說明: 對已經(jīng)加載的模塊進(jìn)行重新加載,一般用于原模塊有變化等特殊情況,reload前該模塊必須已經(jīng)import過,但是需要注意的是已經(jīng)使用的實例還會使用舊模塊,而新產(chǎn)生的實例才會使用新模塊,reload之后還是原來的內(nèi)存地址#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://xmdevops.blog.51cto.com/
#?Purpose:
#
"""
#?說明:?導(dǎo)入公共模塊
import?sys
try:
sys.setdefaultencoding('utf-8')
except?Exception,?e:
print?e
reload(sys)
sys.setdefaultencoding('utf-8')
print?sys.getdefaultencoding()
#?說明:?導(dǎo)入其它模塊
if?__name__?==?'__main__':
pass
說明: 很多人不名為為何要reload()一下sys才能使用setdefaultencoding設(shè)置編碼,其實是因為解釋器初始化時預(yù)先執(zhí)行了/usr/lib64/python2.7/site.py,而在其554行代碼中del sys.setdefaultencoding刪除了此方法,其實你import sys只是指向了那個被刪除了setdefaultencoding屬性的sys模塊地址,所以需要重新reload一下還原此方法
相對導(dǎo)入:
說明: ?PY通過模塊名中的點來判斷是否屬于包,屬于哪個包,當(dāng)你使用from ..xx import oo,其中的點表示包結(jié)構(gòu)中的層次,如果模塊名為__main__表示它不屬于任何包,所以此時模塊名應(yīng)該不包含點,否則會導(dǎo)致relative-import in non-package錯誤,也就是說包含相對導(dǎo)入的文件無法作為入口文件,但是可通過python -m來當(dāng)作模塊載入from?..libs.database?import?Redis
from?..libs.alarm?import?alarm_template
from?..libs.alarm.api?import?weixin_notify
絕對導(dǎo)入:
說明: 絕對導(dǎo)入也叫完全導(dǎo)入,2.x版本必須使用from __future__ import absolute_import打開此機(jī)制,而3.x則將其作為默認(rèn)機(jī)制from?x.y.z?import?o
動態(tài)導(dǎo)入:
說明: __import__其實就是import的內(nèi)部實現(xiàn),通常用于動態(tài)加載,如插件式監(jiān)控系統(tǒng)中只知道插件名如何執(zhí)行插件內(nèi)的代碼?此時就可以通過動態(tài)加載來實現(xiàn)獲取插件內(nèi)的函數(shù)然后去調(diào)用
__import__(module_name[, globals[, locals[, fromlist]]]) ?-> object
說明: module_name為模塊名,但是需要注意的是如果module_name包含子模塊如x.y,則默認(rèn)會返回x對象,如果要返回y對象需要設(shè)置fromlist列表,來實現(xiàn)from x import y的效果.當(dāng)然要實現(xiàn)動態(tài)導(dǎo)入含有專門的imp和importlib模塊.可以學(xué)習(xí)一下~
應(yīng)用場景:
1. zabbix/nagios等監(jiān)控系統(tǒng)主動監(jiān)控都必須手工配置Agent,即使自定義插件亦是如此,如果要實現(xiàn)一個自動檢測插件自動調(diào)用插件自動上報數(shù)據(jù)的監(jiān)控系統(tǒng)要如何實現(xiàn)哪?
#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://my.51CTOBG.net/pydevops/
#?Purpose:
#
"""
#?說明:?兼容絕對導(dǎo)入
from?__future__?import?absolute_import
#?說明:?導(dǎo)入公共模塊
import?json
import?time
import?threading
#?說明:?導(dǎo)入其他模塊
from?..libs.database?import?Redis
from?..libs.alarm?import?alarm_template
from?..libs.alarm.api?import?weixin_notify
#?說明:?客戶端監(jiān)控類
class?MonitorClient(object):
def?__init__(self,?redis,?agent,?info,?error):
self.info?=?info
self.error?=?error
self.redis?=?Redis(db=redis['db'],
host=redis['host'],
port=redis['port'],
password=redis['password'])
self.agent_host?=?agent['host']
self.redis_host?=?redis['host']
self.clientconf?=?self._get_climconf()
self.pubchannel?=?redis['publish']?or?'xmdevops_channel'
self.info.info('update?key#climconf::%s?val#%s'?%?(self.agent_host,?self.clientconf))
def?start(self):
self._plugins_handler()
def?_get_climconf(self):
redis_key?=?'climconf::%s'?%?(self.agent_host,)
while?True:
redis_val?=?self.redis.get(redis_key)
if?not?redis_val:
message?=?'getval?key#%s?with?nothing,5?seconds?try?again'?%?(redis_key,)
self.info.info(message)
self._report_exception(redis_key,?message)
time.sleep(5)
continue
try:
conf_dict?=?json.loads(redis_val)
except?TypeError,?e:
message?=?'unpack?key#%s?val#%s?with?error?%s'?%?(redis_key,?redis_val,?e)
self.error.error(message)
self._report_exception(redis_key,?message)
time.sleep(5)
continue
break
return?conf_dict
def?_plugins_handler(self):
while?True:
for?service_name,?plugin_info?in?self.clientconf.iteritems():
if?len(plugin_info)?
self.clientconf[service_name].append(0)
plugin_name,?check_interval,?add_data,?last_runtime?=?plugin_info
if?time.time()?-?last_runtime?>?check_interval:
self.clientconf[service_name][-1]?=?time.time()
self.info.info('plugin?key#%s?val#%s?is?called'?%?(service_name,?plugin_info))
cur_thread?=?threading.Thread(
target=self._plugins_called,?args=(service_name,?plugin_name,?add_data))
cur_thread.start()
time.sleep(1)
old_clientconf?=?self.clientconf
self.clientconf?=?self._get_climconf()
for?trigger_key,?trigger_val?in?self.clientconf.iteritems():
if?trigger_key?in?old_clientconf:
self.clientconf[trigger_key].append(old_clientconf[trigger_key][-1])
def?_plugins_called(self,?service_name,?plugin_name,?add_data):
plugin_path?=?'app.%s.%s'?%?('plugins',?service_name)
try:
plugin_mods?=?__import__(plugin_path,?fromlist=[service_name])
except?ValueError,?e:
message?=?'import?key#%s?val#%s?with?error?%s'?%?(plugin_path,?e)
self.error.error(message)
self._report_exception(plugin_path,?message)
return
try:
plugin_func?=?getattr(plugin_mods,?plugin_name)
except?AttributeError,?e:
message?=?'plugin?key#%s?val#%s?not?exists'?%?(plugin_mods,?plugin_name)
self.error.error(message)
self._report_exception(plugin_func,?message)
return
plugin_data?=?plugin_func(add_data)
report_data?=?{
'host':?self.agent_host,
'data':?plugin_data,
'service':?service_name
}
data?=?json.dumps(report_data)
self.info.info('publish?key#%s?val#%s'?%?(service_name,?data))
self.redis.publish(self.pubchannel,?data)
def?_report_exception(self,?errors,?details):
message?=?alarm_template?%?(
self.agent_host,?'critical',?errors,
time.strftime('%H:%M:%S',?time.localtime()),?details)
results?=?weixin_notify(message)
if?results:
self.error.error(results)
說明: 如上就是一個自己寫的基于redis的全自動化微型監(jiān)控框架部分核心代碼,首先讀取網(wǎng)頁端下發(fā)下來的監(jiān)控配置,然后利用線程通過__import__動態(tài)調(diào)用插件中的入口監(jiān)控函數(shù),然后將執(zhí)行結(jié)果上報到對應(yīng)區(qū)域的redis,server端再處理閥值數(shù)據(jù)等等,可以作為一個非常好的學(xué)習(xí)案例
總結(jié)
以上是生活随笔為你收集整理的python 运维包_基础入门_Python-模块和包.运维开发中__import__动态导入最佳实践?的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： redis修改端口号后还是占用6379_
- 下一篇： python3层装饰器_python三层