python 运维包_基础入门_Python-模块和包.运维开发中__import__动态导入最佳实践?
常規導入:import?module_name[,module1,...]
from?module_name?import?[*|child[,child1,...]
from?module_name?import?[*|child[,child1,...]?as?alias_name
注意: 導入語句可出現在程序任意位置,自定義包要實現from module_name import *的效果則此模塊必須在__init__.py實現__all__ = ['module_1', 'module_2']
加載一次:
說明: 多次重復使用import語句時,不會重新加載模塊,而是把該模塊的內存地址給引用到本地環境變量==>?x.py?<==
#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://xmdevops.blog.51cto.com/
#?Purpose:
#
"""
#?說明:?導入公共模塊
import?os
#?說明:?導入其它模塊
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:
#
"""
#?說明:?導入公共模塊
import?x
import?os
#?說明:?導入其它模塊
if?__name__?==?'__main__':
print?'os?in?y.py',?id(os)
import?x
重新加載:
說明: 對已經加載的模塊進行重新加載,一般用于原模塊有變化等特殊情況,reload前該模塊必須已經import過,但是需要注意的是已經使用的實例還會使用舊模塊,而新產生的實例才會使用新模塊,reload之后還是原來的內存地址#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://xmdevops.blog.51cto.com/
#?Purpose:
#
"""
#?說明:?導入公共模塊
import?sys
try:
sys.setdefaultencoding('utf-8')
except?Exception,?e:
print?e
reload(sys)
sys.setdefaultencoding('utf-8')
print?sys.getdefaultencoding()
#?說明:?導入其它模塊
if?__name__?==?'__main__':
pass
說明: 很多人不名為為何要reload()一下sys才能使用setdefaultencoding設置編碼,其實是因為解釋器初始化時預先執行了/usr/lib64/python2.7/site.py,而在其554行代碼中del sys.setdefaultencoding刪除了此方法,其實你import sys只是指向了那個被刪除了setdefaultencoding屬性的sys模塊地址,所以需要重新reload一下還原此方法
相對導入:
說明: ?PY通過模塊名中的點來判斷是否屬于包,屬于哪個包,當你使用from ..xx import oo,其中的點表示包結構中的層次,如果模塊名為__main__表示它不屬于任何包,所以此時模塊名應該不包含點,否則會導致relative-import in non-package錯誤,也就是說包含相對導入的文件無法作為入口文件,但是可通過python -m來當作模塊載入from?..libs.database?import?Redis
from?..libs.alarm?import?alarm_template
from?..libs.alarm.api?import?weixin_notify
絕對導入:
說明: 絕對導入也叫完全導入,2.x版本必須使用from __future__ import absolute_import打開此機制,而3.x則將其作為默認機制from?x.y.z?import?o
動態導入:
說明: __import__其實就是import的內部實現,通常用于動態加載,如插件式監控系統中只知道插件名如何執行插件內的代碼?此時就可以通過動態加載來實現獲取插件內的函數然后去調用
__import__(module_name[, globals[, locals[, fromlist]]]) ?-> object
說明: module_name為模塊名,但是需要注意的是如果module_name包含子模塊如x.y,則默認會返回x對象,如果要返回y對象需要設置fromlist列表,來實現from x import y的效果.當然要實現動態導入含有專門的imp和importlib模塊.可以學習一下~
應用場景:
1. zabbix/nagios等監控系統主動監控都必須手工配置Agent,即使自定義插件亦是如此,如果要實現一個自動檢測插件自動調用插件自動上報數據的監控系統要如何實現哪?
#!/usr/bin/env?python
#?-*-?coding:?utf-8?-*-
"""
#
#?Authors:?limanman
#?51CTOBG:?http://my.51CTOBG.net/pydevops/
#?Purpose:
#
"""
#?說明:?兼容絕對導入
from?__future__?import?absolute_import
#?說明:?導入公共模塊
import?json
import?time
import?threading
#?說明:?導入其他模塊
from?..libs.database?import?Redis
from?..libs.alarm?import?alarm_template
from?..libs.alarm.api?import?weixin_notify
#?說明:?客戶端監控類
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的全自動化微型監控框架部分核心代碼,首先讀取網頁端下發下來的監控配置,然后利用線程通過__import__動態調用插件中的入口監控函數,然后將執行結果上報到對應區域的redis,server端再處理閥值數據等等,可以作為一個非常好的學習案例
總結
以上是生活随笔為你收集整理的python 运维包_基础入门_Python-模块和包.运维开发中__import__动态导入最佳实践?的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： redis修改端口号后还是占用6379_
- 下一篇： python3层装饰器_python三层