Database API

JanuaDB

class janua.db.connection.JanuaDB(januadb_file)[source]

Janua database instance

__init__(januadb_file)[source]

Initialize connection to database and create it if it doesn’t exists

Parameters:januadb_file – janua sqlite database file
engine = None

Sqlalchemy engine

session = None

Thread-safe sqlalchemy session

admin = None

Admin database object (AdminDB)

contact = None

Contact database object (ContactDB)

sms = None

Sms database object (SmsDB)

config = None

Config data object (ConfigDB)

action = None

Action database object (ActionDB)

commands = None

Commands database object (CommandsDB)

contact_notify = None

Contact notify action database object (ContactNotifyDB)

contact_group = None

Contact group database object (ContactGroupDB)

authorized_group = None

Authorized group action database object (AuthorizedGroupDB)

group = None

Group database object (GroupDB)

create_db()[source]

Create database if it doesn’t exists

Returns:sqlalchemy engine
create_table(table)[source]

Create a table in Janua database

Parameters:table – table model
add_entry(obj)[source]

Add an object entry

Parameters:obj – entry object to add
Returns:True if entry was added, False otherwise
del_entry(obj)[source]

Delete an object entry

Parameters:obj – entry object to delete
Returns:True if entry was deleted, False otherwise
update_entry(obj)[source]

Update an object entry

Parameters:obj – entry object to update
Returns:True if entry was updated, False otherwise

ObjectDB

class janua.db.connection.ObjectDB(db)[source]

Generic class to connect database APIs to JanuaDB

__init__(db)[source]

Constructor

Parameters:db – JanuaDB instance

ContactDB

class janua.db.connection.ContactDB(db)[source]
add(name, firstname, phone)[source]

Add a contact

Parameters:
  • name – contact name
  • firstname – contact firstname
  • phone_number – contact phone number
Returns:

True if entry was added or False if wasn’t
delete(id)[source]

Delete a contact

Parameters:id – id primary key
Returns:True if entry was deleted or False if wasn’t
get_all()[source]

Get all contact entry

Returns:all Contact entry object
get_by_id(id)[source]

Get a Contact entry corresponding to id

Returns:a Contact entry object
get_by_phone(phone)[source]

Get a Contact entry corresponding to phone number

Returns:a Contact entry object
get_by_group_id(group_id)[source]

Get all contact entries corresponding to group id

Parameters:group_ids – group_id list
Returns:Contact entries object
get_by_admin_phone(phone_number)[source]

Get all contact entry managed by an admin which is identified by his phone number

Parameters:phone_number – admin phone number
Returns:Contact entry object managed by admin
get_by_action_id(action_id)[source]

Get all contact entries affected to an action

Parameters:action_id – action database id
Returns:Contact entries object

GroupDB

class janua.db.connection.GroupDB(db)[source]
get_by_contact_id(contact_id)[source]

Get group entries for contact id

Parameters:contact_id – contact id
Returns:Group entry object
get_by_admin_phone(phone_number)[source]

Get all group entries managed by an admin which is identified by his phone number

Parameters:phone_number – admin phone number
Returns:Group entries object managed by admin

ContactGroupDB

class janua.db.connection.ContactGroupDB(db)[source]
get_by_group_id(group_id)[source]

Get contact group entries

Parameters:group_ids – group id list corresponding to entry to retrieve
Returns:ContactGroup entry object

AdminDB

class janua.db.connection.AdminDB(db)[source]
has_super_admin()[source]

Check if there a superadmin entry in database

Returns:True if yes or False if not
get_super_admin()[source]

Get superadmin entry

Returns:Admin entry object
get_by_phone(phone_number)[source]

Get an admin entry by phone number

Parameters:phone_number – admin phone number
Returns:an Admin entry object
get_by_login(login)[source]

Get an admin entry by login

Parameters:phone_number – admin login
Returns:an Admin entry object
get_by_mail(mail)[source]

Get an admin entry by mail

Parameters:phone_number – admin mail
Returns:an Admin entry object
get_by_id(id)[source]

Get admin entry by id

Parameters:id – admin id
Returns:an Admin entry object
get_all()[source]

Get all admin entry

Returns:all Admin entry object
add(firstname, name, phone_number, password, email, level, login)[source]

Add an admin/supervisor

Parameters:
  • name – admin name
  • firstname – admin firstname
  • phone_number – admin phone number
  • password – hashed sha password
  • email – admin/supervisor email
  • level – 1 for admin, 2 for supervisor
  • login – user login
Returns:

True if entry was added or False if wasn’t
get_level(phone_number)[source]

Get admin level based on phone number aka username credential

Parameters:phone_number – phone number as credential
Returns:admin level or None
get_id_from_phone(phone_number)[source]

Get admin id base on phone number aka username credential

Parameters:phone_number – phone number as credential
Returns:admin id or None

CommandsDB

class janua.db.connection.CommandsDB(db)[source]
get_all()[source]

Get all commands

Returns:Commands entries object

Config

class janua.db.connection.ConfigDB(db)[source]

Config database API

add(key, value)[source]

Add a Config entry object

Parameters:
  • key – configuration
  • value – configuration value
Returns:

True if entry was added or False if wasn’t
delete(key)[source]

Delete a Config entry object

Parameters:key – key entry to delete
Returns:True if entry was deleted or False if wasn’t
get(key)[source]

Get a Config entry object identified by key

Parameters:key – config key entry to get
Returns:Config entry object

ActionDB

class janua.db.connection.ActionDB(db)[source]
add(name, module, description, authentication, enabled, janua_id)[source]

Add an action

Parameters:
  • name – action name
  • module – python module namespace of action
  • description – action description
  • authentication – is action require authentication
  • janua_id – action internal id
Returns:

True if added, False otherwise
get_all()[source]

Get all action entries

Returns:Action entries object
get_by_janua_id(id)[source]

Get action entry by janua action id

Parameters:id – janua internal id (janua_id)
Returns:Action entry object
get_by_authorized_supervisor_id(admin_id)[source]

Get all authorized supervisor action

Parameters:admin_id – admin id
Returns:Action entries object

ContactNotifyDB

class janua.db.connection.ContactNotifyDB(db)[source]
get_by_admin_id(admin_id)[source]

Get all contact notify entries managed by an admin

Parameters:admin_id – admin id
Returns:ContactNotifyAction entries object
get_by_action_id(action_id)[source]

Get contact notify action assigned to an action

Parameters:action_id – action id
Returns:ContactNotifyAction entries

AuthorizedGroupDB

class janua.db.connection.AuthorizedGroupDB(db)[source]
get_by_admin_id(admin_id)[source]

Get all authorized group entries managed by an admin

Parameters:admin_id – admin id
Returns:AuthorizedGroupAction entry object managed by admin

SmsDB

class janua.db.connection.SmsDB(db)[source]
add(date, sender, recipient, message, authorized, admin_id, phone_number, status=3, reference=None, slices=1)[source]

Add a sms log entry

Parameters:
  • date – date time
  • sender – sender address, name or ip address
  • recipient – recipient address or name
  • message – message in raw format (mainly for debug purpose)
  • authorized – is sender authorized to send SMS to server
  • admin_id – admin send message, for received message default to super admin
  • phone_number – recipient phone number
  • status – sms status
  • reference – reference for delivery report
  • slices – number of slices (for multi-part SMS)
Returns:

True if entry was added or False if wasn’t
get_by_admin(admin_id, startdate=None, enddate=None)[source]

Get sms log entries by admin

Parameters:
  • startdate – get entries from date (datetime argument)
  • enddate – get entries to date (datetime argument)
  • admin_id – admin identifier
Returns:

all Log entry object
get_by_date(startdate=None, enddate=None)[source]

Get sms log entries

Parameters:
  • startdate – get entries from date (datetime argument)
  • enddate – get entries to date (datetime argument)
Returns:

all Log entry object
get_by_ref_and_phone(ref, phone_number)[source]

Get sms log entry by recipient phone number and sms reference

Parameters:
  • ref – sms reference
  • phone_number – recipient phone number
Returns:

corresponding sms log entry object
delete(startdate, enddate)[source]

Delete sms log entries in a date interval

Parameters:
  • startdate – delete entries from date (datetime argument)
  • enddate – delete entries to date (datetime argument)
Returns:

all Log entry object
is_admin_quota_reached(admin)[source]

Test if sms send limit has been reached for admin

Parameters:admin – admin entry
Returns:a tuple of 2 value, first value indicate if limit has been reached, second value indicate the number of sms sent
month_usage()[source]

Get sms sent count sent for current month

Returns:number of sms sent
get_admin_month_stats(admin)[source]

Get SMS month statistics usage for an admin

Parameters:admin – admin object
Returns:a list containing dictionary with month/value keys representing number of SMS sent corresponding month