Différences

Ci-dessous, les différences entre deux révisions de la page.

--- python:les_modules [2013/05/31 22:54] – [sqlalchemy] root
+++ python:les_modules [2017/06/03 13:08] (Version actuelle) – [readline] root
@@ Ligne 4: / Ligne 4: @@
   * [[http://docs.python.org/2/library/|Python 2]]
   * [[http://docs.python.org/3/library/|Python 3]]
+===== Chemin des modules =====
+  * Dans l'environnement SHELL:
+<code bash>
+export PYTHONPATH=/home/user/python/lib/mylib
+</code>
+  * Dans le code python :
+<code python>
+#!/usr/bin/python
+import os, sys
+sys.path.append(os.path.join(os.path.dirname(__file__), "lib"))
+#sys.path.insert(0, os.path.join(os.path.dirname(__file__), "lib"))
+</code>
 ===== optparse =====
@@ Ligne 184: / Ligne 198: @@
 #!/usr/bin/env python
-from fabric.api import *
+import sys, os, getpass
-from fabric.colors import *
+#import sys, os, getpass, multiprocessing
+try:
+        from fabric.api import *
+        from fabric.colors import *
+except:
+        print('Fabric module must be installed !')
+        sys.exit(1)
+sys.tracebacklimit = 0
+output['running'] = False
+output['stdout'] = False
+output['aborts'] = False
+output['status'] = False
+output['warnings'] = False
+#env.user = 'root'
+#env.password = None
-env.hosts = ['host1', 'toto@host2']
-env.user = 'root'
-env.password = 'xxx'
-#env.gateway = 'mygateway',
 #env.passwords = {
-    'user@host':'pass',
+#    'user@host':'pass',
-    'gwuser@gwhost':'gwpass'
+#    'gwuser@gwhost':'gwpass'
-}
+#}
 #env.shell = "/bin/sh -c"
+#env.gateway = None
+env.disable_known_hosts = True
+#env.parralel = True
+#env.pool_size = multiprocessing.cpu_count() #Sets the number of concurrent processes to use when executing tasks in parallel.
 #env.keepalive = 300
-#env.roledefs = {
+#env.timeout = 10 #Network ssh connection timeout, in seconds.
-    'web': ['www1', 'www2', 'www3'],
+#env.command_timeout = 60
-    'dns': ['ns1', 'ns2']
+env.linewise = True
+env.warn_only = True
+env.role_default = 'adm'
+env.roledefs = {
+        'eth': ['molx1321', 'molx1322', 'molx1323', 'molx1324', 'molx1325', 'molx1326', 'molx1327', 'molx1328', 'molx1329', 'molx1330', 'molx1331', 'molx1332', 'molx1333', 'molx1334', 'molx1335', 'molx1336' ],
+        'adm' : ['molx1321-adm', 'molx1322-adm', 'molx1323-adm', 'molx1324-adm', 'molx1325-adm', 'molx1326-adm', 'molx1327-adm', 'molx1328-adm', 'molx1329-adm', 'molx1330-adm', 'molx1331-adm', 'molx1332-adm', 'molx1333-adm', 'molx1334-adm', 'molx1335-adm', 'molx1336-adm' ],
+        'ib' : ['molx1321-ib', 'molx1322-ib', 'molx1323-ib', 'molx1324-ib', 'molx1325-ib', 'molx1326-ib', 'molx1327-ib', 'molx1328-ib', 'molx1329-ib', 'molx1330-ib', 'molx1331-ib', 'molx1332-ib', 'molx1333-ib', 'molx1334-ib', 'molx1335-ib', 'molx1336-ib' ],
+        'imm' : ['molx1321-imm', 'molx1322-imm', 'molx1323-imm', 'molx1324-imm', 'molx1325-imm', 'molx1326-imm', 'molx1327-imm', 'molx1328-imm', 'molx1329-imm', 'molx1330-imm', 'molx1331-imm', 'molx1332-imm', 'molx1333-imm', 'molx1334-imm', 'molx1335-imm', 'molx1336-imm' ],
+        'web_adm' : ['sw-eth','sw-ib'],
 }
-env.disable_known_hosts = True
-env.parralel = True
-env.pool_size = 0 #Sets the number of concurrent processes to use when executing tasks in parallel.
-env.timeout = 10 #Network ssh connection timeout, in seconds.
-env.command_timeout = 10
-#env.warn_only = True
-hide('warnings', 'running', 'stdout', 'stderr')
-def remote_info():
+env.roledefs['web_adm'] += env.roledefs['imm']
-        with hide('running', 'stdout'):
-                print(yellow("Executing on %(host)s as %(user)s" % env, bold=True))
-                a = run('uname -a')
-                fastprint(a, show_prefix=True)
-def local_info():
+if env.hosts == [] and env.roles == []:
-        local('uname -a')
+        env.hosts = env.roledefs[env.role_default]
+#list_all = []
+#for server in env.roledefs.values():
+#       list_all += server
+#env.roledefs['all'] = list(set(list_all))
+def VERIF_IPMI_PASSWORD():
+        if 'IPMI_PASSWORD' in os.environ:
+                globals()['IPMI_PASSWORD'] = os.environ['IPMI_PASSWORD']
+        if 'IPMI_PASSWORD' not in globals():
+                IPMI_PASSWORD = getpass.getpass(prompt='IMM password : ')
+                globals()['IPMI_PASSWORD'] = IPMI_PASSWORD
+@task
+@serial
+def lr():
+        """ : Liste les roles utilisables avec l'option '-R'"""
+        for role in env.roledefs:
+                if role == env.role_default:
+                        print("%s (default) : %s" % (role, ' ,'.join(env.roledefs[role])))
+                else:
+                        print("%s : %s" % (role, ' ,'.join(env.roledefs[role])))
+        sys.exit()
+#@parallel(pool_size = multiprocessing.cpu_count() + 1)
+@task
+def cmd(cmd, show = None):
+        """ : Run command on a host (show='details' for details, show='status' for see if command is successful or not)"""
+        #with show('stdout','running','warnings'):
+        try:
+                out = run(cmd)
+        except Exception as err:
+                sys.stdout.write(yellow("%(host)s (%(user)s) : " % env, bold = True))
+                print(red(err))
+                return
+        if show == None:
+                for line in out.splitlines():
+                        sys.stdout.write(yellow("%(host)s (%(user)s) : " % env, bold = True))
+                        print(line)
+                if out.failed:
+                        sys.stdout.write(yellow("%(host)s (%(user)s) : " % env, bold = True))
+                        print(red(' KO', bold = True)) + " (return %s)" % out.return_code
+        elif show == 'details' or show == 'status':
+                sys.stdout.write(yellow("Executing on %(host)s as %(user)s : " % env, bold = True) + blue(cmd, bold = True))
+                if out.failed:
+                        print(red(' KO', bold = True))
+                else:
+                        print(green(' OK', bold = True))
+                if show ==  'details':
+                        print(yellow('OUT : (return code = %s)' % out.return_code))
+                        print('%s\n' % out)
+        else:
+                print("Option %s unknow" % show)
+                sys.exit(1)
+@task
+def send_file(src, dst = None, mode = None):
+        """ : Send file on a host (dst=<destination>, mode=<mode>)"""
+        if dst == None:
+                dst = src
+        if mode == None:
+                put(src, dst)
+        else:
+                try:
+                        int(mode)
+                except:
+                        print("'mode' value must be an integer !")
+                        sys.exit(1)
+                put(src, dst, mode=mode)
+@task
+def reboot(wait = 0):
+        """ : Reboot a host (wait=<secondes>)"""
+        if os.getlogin() != 'root':
+                print('Only root must be reboot a host')
+                sys.exit(1)
+        try:
+                int(wait)
+        except:
+                print("'wait' value must be an integer !")
+                sys.exit(1)
+        reboot(wait = wait)
+@serial
+@task
+def power(state):
+        """ : IMM power manager (state=<status|on|off|cycle|reset|diag|soft>)"""
+        if not env.host in env.roledefs['imm']:
+                print('Use "-R imm" or "-H <host>-imm"')
+                print('%s not in this table : %s' % (env.host, ','.join(env.roledefs['imm'])))
+                sys.exit(1)
+        if state not in ['status','on','off','cycle','reset','diag','soft']:
+                print("%s state is not supported. Please use state <status|on|off|cycle|reset|diag|soft> !" % state)
+                sys.exit(1)
+        VERIF_IPMI_PASSWORD()
+        out = local("ipmitool -I lanplus -C 2 -U system -P '%s' -E -H %s chassis power %s" % (globals()['IPMI_PASSWORD'], env.host, state), capture=True)
+        if out.failed:
+                sys.stdout.write(yellow("%(host)s : " % env, bold = True))
+                print(red('Failed ! (return code : %s)' % out.return_code))
+                print(out.stderr)
+                sys.exit(1)
+        else:
+                for line in out.splitlines():
+                        sys.stdout.write(yellow("%(host)s : " % env, bold = True))
+                        print(line)
+@serial
+@task
+def console(action='activate'):
+        """ : IMM SOL (action=<activate|deactivate>)"""
+        if len(env.hosts) != 1:
+                print('Only on one host. Use option -H to precise the host !')
+                sys.exit(1)
+        if not env.host in env.roledefs['imm']:
+                print('Use "-H <host>-imm"')
+                print('%s not in this table : %s' % (env.host, ','.join(env.roledefs['imm'])))
+                sys.exit(1)
+        VERIF_IPMI_PASSWORD()
+        os.system("ipmitool -I lanplus -C 2 -U system -P '%s' -E -H %s sol %s" % (globals()['IPMI_PASSWORD'], env.host, action))
+@serial
+@task
+def web_adm():
+        """ : Open Firefox to administrate (use -H option)"""
+        if len(env.hosts) != 1:
+                print('Only on one host. Use option -H to precise the host (%s)!' % ', '.join(env.roledefs['web_adm']))
+                sys.exit(1)
+        if os.getenv('DISPLAY') == None:
+                print('You must have a valid display !')
+                sys.exit(1)
+        if env.host not in env.roledefs['web_adm']:
+                print("%s is not a valid host ! Valid host are %s" % (env.host, ', '.join(env.roledefs['web_adm'])))
+                sys.exit(1)
+        os.system('firefox https://%s/' % env.host)
 </code>
+===== readline =====
+<code python readline.py>
+import readline, signal
+prompt = 'GigiX > '
+def signal_handler(signal, frame):
+    print
+    input(prompt)
+    return
+signal.signal(signal.SIGINT, signal_handler)
+readline.parse_and_bind("tab: complete")
+class completer:
+        def __init__(self, tree):
+            self.tree = tree
+        def _parser(self, line, search):
+            if (search is None) or (callable(search) == True):
+                return []
+            elif len(line) == 0:
+                return [x for x in search]
+            elif len(line) == 1 and readline.get_line_buffer()[-1] != ' ':
+                return ['%s ' % x for x in search if x.startswith(line[0])]
+            else:
+                return self._parser(line[1:], search[line[0]])
+            return []
+        def complete(self, text, state):
+            line = readline.get_line_buffer().split()
+            results = self._parser(line, self.tree)
+            return results[state]
+def linux_sles_12():
+    print('Je suis Linux SLES 12')
+tree = {
+    'unix': {
+        'solaris': {
+            '8':None,
+            '9':None,
+            '10':None,
+            '11':None,
+        }
+    },
+    'linux': {
+        'debian': {
+            'buzz':None,
+            'rex':None,
+            'bo':None,
+            'hamm':None,
+            'slink':None,
+            'patato':None,
+            'woody':None,
+            'sarge':None,
+            'etch':None,
+            'lenny':None,
+            'squeeze':None,
+            'wheezy':None,
+            'jessie':None,
+    },
+        'RHEL': {
+            '3':None,
+            '4':None,
+            '5':None,
+            '6':None,
+            '7':None,
+        },
+        'SLES': {
+            '9':None,
+            '10':None,
+            '11':None,
+            '12':linux_sles_12,
+        }
+    },
+    'windows': {
+        'xp':None,
+        'vista':None,
+        'seven':None,
+        '8':None,
+        '10':None,
+    }
+}
+readline.set_completer(completer(tree).complete)
+l = ''
+while l != ('quit' or 'exit'):
+    try:
+            l = input(prompt)
+            eval("tree['"+"']['".join(l.split())+"']()")
+    except EOFError:
+            print('Bye')
+            break
+    except:
+        continue
+</code>
 ===== Twisted =====
@@ Ligne 245: / Ligne 527: @@
 import ctypes, ctypes.util
-ctypes.cdll.LoadLibrary(ctypes.util.find_library('c'))
+libc=ctypes.cdll.LoadLibrary(ctypes.util.find_library('c'))
 libc.printf("uid = %d - gid = %d\n", libc.geteuid(), libc.getgid())
 </code>
@@ Ligne 461: / Ligne 743: @@
 </code>
-  * groupby
+  * groupby :
 <code python>
@@ Ligne 478: / Ligne 760: @@
 </code>
+  * izip :
+<code python>
+>>> for i in itertools.izip([1, 2, 3], ['a', 'b', 'c']):
+...  i
+...
+(1, 'a')
+(2, 'b')
+(3, 'c')
+</code>
+<code python>
+>>> for i in itertools.izip(itertools.count(5), ['a', 'b', 'c']): i
+...
+(5, 'a')
+(6, 'b')
+(7, 'c')
+</code>
+  * cycle :
+<code python>
+>>> for i in itertools.cycle(['a', 'b', 'c']):i
+'a'
+'b'
+'c'
+'a'
+'b'
+'c'
+'a'
+'b'
+'c'
+'a'
+'b'
+'c'
+...
+...
+</code>
 ===== string =====
@@ Ligne 491: / Ligne 810: @@
 >>> string.whitespace
 '\t\n\x0b\x0c\r '
+</code>
+===== zip =====
+<code python>
+>>> zip(['a', 'b', 'c'], ['d', 'e', 'f'])
+[('a', 'd'), ('b', 'e'), ('c', 'f')]
+</code>
+===== operator =====
+Trie par rapport a un index :
+<code python>
+>>> import operator
+>>> d = dict(a=1, b=2, c=1, d=2, e=1, f=2, g=3)
+>>> sorted(d.iteritems(), key=operator.itemgetter(1))
+[('a', 1), ('c', 1), ('e', 1), ('b', 2), ('d', 2), ('f', 2), ('g', 3)]
+</code>
+===== dbus =====
+Dbus permet d’interagir avec les autres programmes identifiés par DBUS comme knotify, le service de notifications KDE.
+Pour voir la liste des services DBUS, lancer la commande qdbus :
+<code>
+gigi@debian:~$ qdbus| grep knotify
+ org.kde.knotify
+</code>
+<code python knotify.py>
+#!/usr/bin/env python
+import sys, dbus
+knotify = dbus.SessionBus().get_object("org.kde.knotify", "/Notify")
+try:
+    title, text = sys.argv[1:3]
+except:
+    print 'Usage: knotify.py title text'
+    sys.exit(1)
+knotify.event("warning", "kde", [], title, text, [], [], 0, 0,dbus_interface="org.kde.KNotify")
 </code>