Demo Library kopdar python-ID Jogja 240916 untuk berinteraksi dengan zimbra API dengan dependensi module requests

ozsoap.py

__author__ = ('Imam Omar Mochtar', ('omar@jabetto.com', 'iomarmochtar@gmail.com'))
__version__ = (0 ,4)


"""
Dokumentasi zimbra soap berdasarkan versinyas
http://wiki.zimbra.com/wiki/SOAP_API_Reference_Material_Beginning_with_ZCS_8.0
"""

"""
Changelogs
==========
0.4:
- Management COS (getAll, create, delete, modify)
- Disable warnings https
- menghapus account (DeleteAccount)

0.3:
- pada ZMAIL otomatis set iszadmin menjadi false

0.2:
- memambahkan exception 
- menambahkan ca => CreateAccount

0.1:
- Mengubah menjadi zimbra lib 
- Menambahkan class ZMAIL untuk menerima dan mengirimkan email
"""

import requests
import json
import os
from pprint import pprint
from ConfigParser import ConfigParser

requests.packages.urllib3.disable_warnings()

class ZLoginFailed(Exception): pass
class ZConnectionErr(Exception): pass
class ZCommonErr(Exception): pass

class JZSOAP(object):
	"""
	Base SOAP
	"""
	token = None
	soapurl = None
	isdebug = False

	username = None
	password = None

	def __init__(self, username=None, password=None, soapurl=None, isdebug=False, iszadmin=True, timeout=None, zinstance=None):

		if not zinstance and None in (username, password, soapurl):
			raise ZCommonErr("You must specified username, password and soapurl or zinstance to initiate object")

		self.isdebug = isdebug
		self.soapurl = soapurl
		self.iszadmin = iszadmin
		self.timeout = timeout

		# jika ada instance dari zobjek maka set attribute saja
		if zinstance:
			self.setAttrs(zinstance)
		else:
			self.__login(username, password)


	def __login(self, username, password):
		"""
		Login untuk mendapatkan token
		"""
		soapname = "Auth"
		if self.iszadmin:
			urn = "urn:zimbraAdmin"
			body = {
				"name": [{
					"_content": username
				}],
				"password": [{
					"_content": password
				}]
			}
		else:
			urn = "urn:zimbraAccount"
			body = {
				"account": {
					"by": "name",
					"_content": username
				},
				"password": {
					"_content": password
				}
			}

		result = self.send(soapname, body, urn)

		if not result:
			raise ZLoginFailed("Login Failed")

		self.token = result['authToken'][0]['_content']

	def getAttrs(self):
		"""
		Mendapatkan attribute yang akan didapatkan oleh objek
		"""
		return dict(
			token = self.token,
			soapurl = self.soapurl,
			isdebug = self.isdebug,
			username = self.username,
			password = self.password
		)

	def setAttrs(self, zinstance):
		"""
		Set attribute dari zinstace yang diberikan pada current objek
		"""
		attrs = zinstace.getAttrs()
		for k,v in attrs.iteritems():
			setattr(self, k, v)


	def printMe(self, word, isarray=False):
		"""
		Cetak verbose ke konsole
		"""
		if not self.isdebug: return

		if not isarray:
			print word
		else:
			pprint(word)


	def send(self, soapname, body={}, urn="urn:zimbraAdmin"):
		"""
		Mengirimkan data dalam bentuk JSON, dan menerima dalam bentuk json
		"""

		# tambahkan urn di body
		body['_jsns'] = urn

		sendjson = {
			"Header": {
				"context": {
					"authToken": [{'_content':self.token} if self.token else {}],
					"nosession": {},
					"userAgent": {
						"name": "zmsoap"
					},
					"_jsns": "urn:zimbra"
				}
			},
			"Body": {"%sRequest"%soapname:body},
			"_jsns": "urn:zimbraSoap"
		}

		self.printMe('\nRequest BEGIN\n')
		self.printMe(sendjson, isarray=True)
		self.printMe('\nRequest END\n')

		kwargs = {'verify': False}
		if self.timeout:
			kwargs['timeout'] = self.timeout

		try:
			result = requests.post(self.soapurl, json.dumps(sendjson), **kwargs).json()
		except requests.exceptions.ConnectionError, e:
			raise ZConnectionErr("Connection error: {0}".format(e))

		body = result['Body']
		# jika terjadi kesalahan
		if body.has_key('Fault'):
			self.printMe(body, True)
			return False

		# balikan sesuai dengan response yang diminta
		self.printMe('\nResponse BEGIN\n')
		self.printMe(body, isarray=True)
		self.printMe('\nResponse END\n')
		balik = body['%sResponse'%soapname]
		return balik


class ZMAIL(JZSOAP):

	"""
	Mengirimkan dan menerima email (mailbox punya).
	"""


	def __init__(self, **kwargs):

		if kwargs.has_key('iszadmin'):
			del kwargs['iszadmin']

		JZSOAP.__init__(self, iszadmin=False, **kwargs)

	def sendMail(self, to, subject, body):
		"""
		Mengirimkan email
		"""
		soapname = "SendMsg"
		urn = "urn:zimbraMail"
		body = {
			"m":{
				"e":[
					{
						"t": "t",
						"a": to,
						"add": "0"
					},
				],
				"su":{
					"_content": subject
				},
				"mp":[
					{
						"ct":"text/plain",
						"content":{
							"_content": body
						}
					}
				]
			}
		}
		return self.send(soapname, body, urn)

	def getMailbox(self, query="in:\"inbox\"", offset=0, limit=1):
		"""
		Mendapatkan isi mailbox
		"""
		soapname = "Search"
		urn = "urn:zimbraMail"
		body = {
			"sortBy": "dateDesc",
			"tz": {
				"id": "Asia/Bangkok"
			},
			"locale": {
				"_content": "en_US"
			},
			"offset": offset,
			"limit": limit,
			"query": query,
			"types": "conversation",
			"fetch": 1,
			"html": 1
		}

		return self.send(soapname, body, urn)



class ZMPROV(JZSOAP):

	"""
	wrapper untuk zmprov command line yang menggunakan soap
	"""

	def gadl(self):
		"""
		Wrap seakan-akan menggunakan zmprov gadl
		"""
		exclude_dist = ['zimbradladmins@z.ong', 'zimbradomainadmins@z.ong']
		

		head = 'GetAllDistributionLists'
		data = {} 

		result = self.send(head, data)
		return [x['name'] for x in result['dl'] if x['name'] not in exclude_dist ]

	"""
	# gdlm(distname, searchdata['search'], searchdata['page'])
	def gdlm(self, distname, search, page):


		head = 'GetDistributionList'
		data = { 
			"dl": [{
				"by": "name",
				"_content": distname
			}]
		}  
		urn = "urn:zimbraAdmin"

		members = []
		result = self.send(head, data, urn)

		try:
			dlmlist = result['dl'][0]['dlm']
			return paginate(dlmlist, '_content', search, page)
			# return [ x['_content'] for x in result['dl'][0]['dlm'] ]
		except:
			return None
	"""

	def rdlm(self, distid, members):
		"""
		Menghapus distribution list member
		"""

		head = 'RemoveDistributionListMember'

		data = {
			'id': distid,
			'dlm': [{'_content':x} for x in members ]
		}

		return self.send(head, data)

	def adlm(self, distid, members):
		"""
		Menambahkan distribution list member
		"""

		head = 'AddDistributionListMember'

		data = {
			'id': distid,
			'dlm': [{'_content':x} for x in members ]
		}

		return self.send(head, data)	

	def ra(self, accountid, newname):
		"""
		Rename account
		"""
		data = {
			'id': accountid,
			'newName': newname	
		}

		return self.send('RenameAccount', data)

	def ma(self, accountid, attribute, value=None):
		"""
		Modifikasi attribute user
		"""
		data = {
			'id': accountid,
			'a':{'n': attribute, '_content': value}
		}

		# jika tidak ada value berarti akan dihapus
		if not value: del data['a']['_content']

		return self.send('ModifyAccount', data)
	
	def aaa(self, accountid, newalias):
		"""
		Menambahkan alias
		"""
		data = {
			'id': accountid,
			'alias': newalias	
		}

		return self.send('AddAccountAlias', data)


	def raa(self, accountid, alias):
		"""
		Menghapus account alias
		"""
		data = {
			"id": accountid,
			"alias": alias	
		}

		return self.send('RemoveAccountAlias', data)

	def gdlcheck(self, distname):
		"""
		Get distribution list id
		"""

		head = 'GetDistributionList'
		data = { 
			"dl": [{
				"by": "name",
				"_content": distname
			}]
		} 

		result = self.send(head, data)

		if result:
			return result['dl'][0]['id']

		return False

	def ga(self, account):
		"""
		get account
		"""

		head = 'GetAccount'
		data = {
			'account': {
				'by': 'name',	
				"_content": account
			}
		}

		result = self.send(head, data)

		if result:
			return result['account'][0]['a']

		return False


	def cdl(self, distname):
		"""
		Membuat distribution list baru
		"""

		head = 'CreateDistributionList'
		data = {
			'name': distname,
			'a': {
				"n": "zimbraMailStatus",
			}
		}

		result = self.send(head, data)

		if result:
			return True

		return False

	def ddl(self, distname):
		"""
		Menghapus distribution list
		"""

		# dapatkan dulu id dari distribution list yang dimaksud
		dlid = self.gdlcheck(distname)

		if not dlid:
			return False

		head = 'DeleteDistributionList'
		data = {
			'id': dlid
		}

		result = self.send(head, data)

		if result:
			return True

		return False

	def ca(self, name, password, attrs=[]):
		"""
		create account
		"""
		head = "CreateAccount"
		data = {
			"name": name,
			"password": password,
			"a": attrs	
		}
		return self.send(head, data)

	def da(self, name):
		"""
		Menhapus user
		"""

		acc = self.ga(name)
		if not acc:
			raise ZCommonErr("User {0} not found".format(name))

		zimbraId = None
		for data in acc:
			if data['n'] != 'zimbraId': continue
			zimbraId = data['_content']
			break

		head = "DeleteAccount"
		data = {
			"id": zimbraId
		}

		return self.send(head, data)

	def gac(self, shortd=False):
		"""
		Mendapatkan semua cos yang ada
		@shortd: Tampilkan hanya data sekilas saja
		"""
		result = self.send("GetAllCos")
		if not shortd: return result
		take = []
		for x in result['cos']:
			del x['a']
			take.append( x )
		return take

	def mc(self, cosid, attr, value):
		"""
		Modify COS
		"""

		data = {
			'id': [{
				'_content': cosid
			}],
			'a':{'n': attr, '_content': value}
		}

		return self.send("ModifyCos", data)


	def gc(self, cosid, attrs=None):
		"""
		Mendapatkan COS berdasarkan ID yang dimasukan
		"""
		params = {
			"cos": {
					"by": "id",
					"_content": cosid
				} 
			}
		if attrs:
			params["attrs"] = attrs

		return self.send("GetCos", params)


	# untuk getAllAccount berdasarkan name
	def gaa(self, search='', uid=False, limit=0, offset=0):
		"""
		Mendapatkan semua user
		"""

		head = 'SearchDirectory'
		data = {
			'sortBy': 'name',
			'limit': limit,
			'offset' : offset,
			'sortAscending': 1,
			'attrs' : '',
			# hanya mendapatkan normal account
			"query": "(&(!(zimbraIsSystemAccount=TRUE)))"
		}

		if search:
			data['query'] = 'mail=*{0}*'.format(search)

		members = []

		result = self.send(head, data)

		if not result:
			return []

		if result.has_key('account'):
			for member in result['account']:
				if uid:
					data = {
						'name': member['name'],
						'uid': member['id']	
					}
				else:
					data = member['name']

				members.append(data)

		return members

		# untuk prevpage

		# if page * settings.TOTAL_CONTENT_PER_PAGE > settings.TOTAL_CONTENT_PER_PAGE:
		# 	prevpage = False
		# else:
		# 	prevpage = True

		# nextpage = not result['more'] 

		# if result['searchTotal'] == 0 or len(members) == 0:
		# 	prevpage = True
		# 	nextpage = True

		# return {'members': members, 'nextpage': nextpage, 'prevpage': prevpage }




class ZMailBox(JZSOAP):

	def getMbox(self, mboxid):
		"""
		Mendapatkan mailbox suatu akun
		"""
		return self.send(head, data)


if __name__ == '__main__':
	zmprov = ZMPROV(
	 	username='admin',
	 	password='passwd123',
	 	soapurl='https://127.0.0.1:7071/service/admin/soap',
	 	# isdebug=True
	)

	pprint( zmprov.da("omar@mail.local")  )

i have create it's own repository for further development.

https://github.com/iomarmochtar/ozpy