[refactor] copy DAOs from xivo-ctid and xivo-stat to xivo-dao

parent 639f0a70
# -*- coding: utf-8 -*-
# vim: set fileencoding=utf-8 :
# XiVO CTI Server
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.agentfeatures import AgentFeatures
from xivo_dao.alchemy import dbconnection
class AgentFeaturesDAO(object):
def __init__(self, session):
self._session = session
def agent_number(self, agentid):
return self._get_one(agentid).number
def agent_context(self, agentid):
return self._get_one(agentid).context
def agent_ackcall(self, agentid):
return self._get_one(agentid).ackcall
def agent_interface(self, agentid):
try:
return 'Agent/%s' % self._get_one(agentid).number
except LookupError:
return None
def agent_id(self, agent_number):
if agent_number == None:
raise ValueError('Agent number is None')
res = self._session.query(AgentFeatures).filter(AgentFeatures.number == agent_number)
if res.count() == 0:
raise LookupError('No such agent')
agent_id = res[0].id
return str(agent_id)
def _get_one(self, agentid):
# field id != field agentid used only for joining with staticagent table.
if agentid == None:
raise ValueError('Agent ID is None')
res = self._session.query(AgentFeatures).filter(AgentFeatures.id == int(agentid))
if res.count() == 0:
raise LookupError('No such agent')
return res[0]
@classmethod
def new_from_uri(cls, uri):
connection = dbconnection.get_connection(uri)
return cls(connection.get_session())
# vim: set fileencoding=utf-8 :
# XiVO CTI Server
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String, Text
class AgentFeatures(Base):
__tablename__ = 'agentfeatures'
id = Column(Integer, primary_key=True)
numgroup = Column(Integer, nullable=False)
firstname = Column(String(128), nullable=False, server_default='')
lastname = Column(String(128), nullable=False, server_default='')
number = Column(String(40), nullable=False)
passwd = Column(String(128), nullable=False)
context = Column(String(39), nullable=False)
language = Column(String(20), nullable=False)
silent = Column(Integer, nullable=False, default=0)
autologoff = Column(Integer)
ackcall = Column(String(20), nullable=False, server_default='no')
acceptdtmf = Column(String(1), nullable=False, server_default='#')
enddtmf = Column(String(1), nullable=False, server_default='*')
wrapuptime = Column(Integer)
musiconhold = Column(String(80))
group = Column(String(255))
commented = Column(Integer, nullable=False, default=0)
description = Column(Text)
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()
Type = declarative_base()
# -*- coding: UTF-8 -*-
from sqlalchemy.schema import Column
from sqlalchemy.types import DateTime, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from xivo_dao.alchemy.base import Base
class CEL(Base):
__tablename__ = 'cel'
id = Column(Integer, primary_key=True)
eventtype = Column(String(30), nullable=False)
eventtime = Column(DateTime, nullable=False)
userdeftype = Column(String(255), nullable=False)
cid_name = Column(String(80, convert_unicode=True), nullable=False)
cid_num = Column(String(80, convert_unicode=True), nullable=False)
cid_ani = Column(String(80), nullable=False)
cid_rdnis = Column(String(80), nullable=False)
cid_dnid = Column(String(80), nullable=False)
exten = Column(String(80, convert_unicode=True), nullable=False)
context = Column(String(80), nullable=False)
channame = Column(String(80, convert_unicode=True), nullable=False)
appname = Column(String(80), nullable=False)
appdata = Column(String(80), nullable=False)
amaflags = Column(Integer, nullable=False)
accountcode = Column(String(20), nullable=False)
peeraccount = Column(String(20), nullable=False)
uniqueid = Column(String(150), nullable=False)
linkedid = Column(String(150), nullable=False)
userfield = Column(String(255), nullable=False)
peer = Column(String(80), nullable=False)
#!/usr/bin/python
# vim: set fileencoding=utf-8 :
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
class ContextInclude(Base):
__tablename__ = 'contextinclude'
context = Column(String(39), primary_key=True)
include = Column(String(39), primary_key=True)
priority = Column(Integer, nullable=False, default=0)
# -*- coding: utf-8 -*-
from sqlalchemy.engine import create_engine
from sqlalchemy.orm.session import sessionmaker
_DB_CONNECTION_POOL = None
class DBConnection(object):
def __init__(self, uri):
self._engine = None
self._session = None
if '?' in uri:
self._uri = uri.split('?')[0]
else:
self._uri = uri
def connect(self):
self._engine = create_engine(self._uri)
Session = sessionmaker(bind=self._engine)
self._session = Session()
def close(self):
if self._session is not None:
self._session.close()
self._session = None
def get_engine(self):
return self._engine
def get_session(self):
return self._session
class DBConnectionPool(object):
def __init__(self, connection_factory):
self._connection_factory = connection_factory
self._key_to_connection = {}
self._uri_to_connection = {}
def close(self):
for connection in self._uri_to_connection.itervalues():
connection.close()
self._key_to_connection = {}
self._uri_to_connection = {}
def _do_add_connection(self, uri, key):
if key is None:
key = uri
if uri not in self._uri_to_connection:
connection = self._connection_factory(uri)
connection.connect()
self._uri_to_connection[uri] = connection
if key not in self._key_to_connection:
self._key_to_connection[key] = self._uri_to_connection[uri]
def add_connection(self, uri):
self._do_add_connection(uri, None)
def add_connection_as(self, uri, key):
self._do_add_connection(uri, key)
def get_connection(self, uri):
return self._key_to_connection[uri]
def register_db_connection_pool(db_connection_pool):
global _DB_CONNECTION_POOL
_DB_CONNECTION_POOL = db_connection_pool
def unregister_db_connection_pool():
global _DB_CONNECTION_POOL
if _DB_CONNECTION_POOL is not None:
_DB_CONNECTION_POOL.close()
_DB_CONNECTION_POOL = None
def get_db_connection_pool():
return _DB_CONNECTION_POOL
def _get_connection_pool_or_raise():
db_connection_pool = _DB_CONNECTION_POOL
if db_connection_pool is None:
raise Exception("No registered DB connection pool")
else:
return db_connection_pool
def add_connection(uri):
db_connection_pool = _get_connection_pool_or_raise()
db_connection_pool.add_connection(uri)
def add_connection_as(uri, key):
db_connection_pool = _get_connection_pool_or_raise()
db_connection_pool.add_connection_as(uri, key)
def get_connection(uri):
db_connection_pool = _get_connection_pool_or_raise()
return db_connection_pool.get_connection(uri)
#!/usr/bin/python
# vim: set fileencoding=utf-8 :
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
class Extension(Base):
__tablename__ = 'extensions'
id = Column(Integer, primary_key=True)
commented = Column(Integer)
context = Column(String(39), nullable=False, server_default='')
exten = Column(String(40), nullable=False, server_default='')
priority = Column(Integer, nullable=False, default=0)
app = Column(String(128), nullable=False, server_default='')
appdata = Column(String(128), nullable=False, server_default='')
name = Column(String(128), nullable=False, server_default='')
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall. See the LICENSE file at top of the source tree
# or delivered in the installable package in which XiVO CTI Server is
# distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
class GroupFeatures(Base):
__tablename__ = 'groupfeatures'
id = Column(Integer, primary_key=True)
name = Column(String(128), nullable=False)
number = Column(String(40), nullable=False, default='')
context = Column(String(39))
transfer_user = Column(Integer, nullable=False, default=0)
transfer_call = Column(Integer, nullable=False, default=0)
write_caller = Column(Integer, nullable=False, default=0)
write_calling = Column(Integer, nullable=False, default=0)
timeout = Column(Integer, nullable=False, default=0)
preprocess_subroutine = Column(String(39))
deleted = Column(Integer, nullable=False, default=0)
# vim: set fileencoding=utf-8 :
# XiVO CTI Server
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.types import Integer, String, Text, Enum
from sqlalchemy.schema import Column
class LineFeatures(Base):
__tablename__ = 'linefeatures'
id = Column(Integer, primary_key=True)
protocolid = Column(Integer, nullable=False)
iduserfeatures = Column(Integer, default=0)
config = Column(String(128))
device = Column(String(32))
configregistrar = Column(String(128))
name = Column(String(128), nullable=False)
number = Column(String(40))
context = Column(String(39), nullable=False)
provisioningid = Column(Integer, nullable=False)
rules_type = Column(String(16))
rules_time = Column(String(8))
rules_order = Column(Integer, default=0)
rules_group = Column(String(16))
num = Column(Integer, default=0)
line_num = Column(Integer, default=0)
ipfrom = Column(String(15))
internal = Column(Integer, nullable=False, default=0)
commented = Column(Integer, nullable=False, default=0)
description = Column(Text)
protocol = Column(String(10))
#!/usr/bin/python
# vim: set fileencoding=utf-8 :
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String, Text
class MeetmeFeatures(Base):
__tablename__ = 'meetmefeatures'
id = Column(Integer, primary_key=True)
meetmeid = Column(Integer, nullable=False)
name = Column(String(80), nullable=False)
confno = Column(String(40), nullable=False)
context = Column(String(39), nullable=False, server_default='')
admin_typefrom = Column(String(40)) # Should be Enum ('none', 'internal', 'external', 'undefined')
admin_internalid = Column(Integer)
admin_externalid = Column(String(40))
admin_identification = Column(String(40)) # Should be Enum ('calleridnum', 'pin', 'all')
admin_mode = Column(String(40)) # Should be Enum ('listen', 'talk', 'all')
admin_announceusercount = Column(Integer, nullable=False, default=0)
admin_announcejoinleave = Column(String(40)) # Should be Enum ('no', 'yes', 'noreview')
admin_moderationmode = Column(Integer, nullable=False, default=0)
admin_initiallymuted = Column(Integer, nullable=False, default=0)
admin_musiconhold = Column(String(128))
admin_poundexit = Column(Integer, nullable=False, default=0)
admin_quiet = Column(Integer, nullable=False, default=0)
admin_starmenu = Column(Integer, nullable=False, default=0)
admin_closeconflastmarkedexit = Column(Integer, nullable=False, default=0)
admin_enableexitcontext = Column(Integer, nullable=False, default=0)
admin_exitcontext = Column(String(39))
user_mode = Column(String(40)) # Should be Enum ('listen', 'talk', 'all')
user_announceusercount = Column(Integer, nullable=False, default=0)
user_hiddencalls = Column(Integer, nullable=False, default=0)
user_announcejoinleave = Column(String(40)) # Should be Enum ('no', 'yes', 'noreview')
user_initiallymuted = Column(Integer, nullable=False, default=0)
user_musiconhold = Column(String(128))
user_poundexit = Column(Integer, nullable=False, default=0)
user_quiet = Column(Integer, nullable=False, default=0)
user_starmenu = Column(Integer, nullable=False, default=0)
user_enableexitcontext = Column(Integer, nullable=False, default=0)
user_exitcontext = Column(String(39))
talkeroptimization = Column(Integer, nullable=False, default=0)
record = Column(Integer, nullable=False, default=0)
talkerdetection = Column(Integer, nullable=False, default=0)
noplaymsgfirstenter = Column(Integer, nullable=False, default=0)
durationm = Column(Integer)
closeconfdurationexceeded = Column(Integer, nullable=False, default=0)
nbuserstartdeductduration = Column(Integer)
timeannounceclose = Column(Integer)
maxusers = Column(Integer, nullable=False, default=0)
startdate = Column(Integer)
emailfrom = Column(String(255))
emailfromname = Column(String(255))
emailsubject = Column(String(255))
emailbody = Column(Text)
preprocess_subroutine = Column(String(39))
description = Column(Text)
commented = Column(Integer, default=0)
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
from xivo_dao.alchemy.base import Base
class PhoneFunckey(Base):
__tablename__ = 'phonefunckey'
iduserfeatures = Column(Integer, nullable=False, primary_key=True)
fknum = Column(Integer, nullable=False, primary_key=True)
exten = Column(String(40))
typevalextenumbers = Column(String(255))
typevalextenumbersright = Column(String(255))
label = Column(String(32))
typeextenumbers = Column(String(16))
supervision = Column(Integer, nullable=False, default=0)
progfunckey = Column(Integer, nullable=False, default=0)
typeextenumbersright = Column(String(16))
# -*- coding: UTF-8 -*-
# Copyright (C) 2012 Avencall
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall SAS. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import String
class QueueLog(Base):
__tablename__ = 'queue_log'
time = Column(String(26), nullable=False, server_default='', primary_key=True)
callid = Column(String(32), nullable=False, server_default='', primary_key=True)
queuename = Column(String(50), nullable=False, server_default='')
agent = Column(String(50), nullable=False, server_default='')
event = Column(String(20), nullable=False, server_default='')
data1 = Column(String(30), server_default='')
data2 = Column(String(30), server_default='')
data3 = Column(String(30), server_default='')
data4 = Column(String(30), server_default='')
data5 = Column(String(30), server_default='')
# -*- coding: utf-8 -*-
# Copyright (C) 2012 Avencall
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Avencall SAS. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
class QueuePeriodicStat(object):
pass
# vim: set fileencoding=utf-8 :
# XiVO CTI Server
# Copyright (C) 2007-2012 Avencall
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# Alternatively, XiVO CTI Server is available under other licenses directly
# contracted with Pro-formatique SARL. See the LICENSE file at top of the
# source tree or delivered in the installable package in which XiVO CTI Server
# is distributed for more details.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION
class QueueFeatures(Base):
__tablename__ = 'queuefeatures'
id = Column(Integer, primary_key=True)
name = Column(String(128), nullable=False)
displayname = Column(String(128), nullable=False)
number = Column(String(40), nullable=False, default='')
context = Column(String(39))
data_quality = Column(Integer, nullable=False, default=0)
hitting_callee = Column(Integer, nullable=False, default=0)
hitting_caller = Column(Integer, nullable=False, default=0)
retries = Column(Integer, nullable=False, default=0)
ring = Column(Integer, nullable=False, default=0)
transfer_user = Column(Integer, nullable=False, default=0)
transfer_call = Column(Integer, nullable=False, default=0)
write_caller = Column(Integer, nullable=False, default=0)
write_calling = Column(Integer, nullable=False, default=0)
url = Column(String(255), nullable=False, default='')
announceoverride = Column(String(128), nullable=False, default='')
timeout = Column(Integer, nullable=False, default=0)
preprocess_subroutine = Column(String(39))
announce_holdtime = Column(Integer, nullable=False, default=0)
waittime = Column(Integer)
waitratio = Column(DOUBLE_PRECISION)
# -*- coding: UTF-8 -*-
from xivo_dao.alchemy.base import Base
from sqlalchemy.schema import Column
from sqlalchemy.types import Integer, String
class QueueInfo(Base):
__tablename__ = 'queue_info'
id = Column(Integer, primary_key=True)
call_time_t = Column(Integer)
</