This is an assignment presented to me. I don't have much experience building servers in Python so any comments would help me improve. The point of the server is to save and retrieve messages in a memory expensive environment.

Those are the constraints presented to me: You don’t need to implement pagination

  • Write everything from scratch: no frameworks
  • What we are interested in is how you are going to search those messages.
  • You cannot use a database.
  • Bonus for full-text search on the message bodies
  • Memory is expensive.
  • Keep it simple.

For my solution I opted to store and read the messages on a filesystem. I also use Trie tree for indexing. I however received Email saying that there are serious feedback on the matter. Would love to learn from my mistakes, but unless I know them, its hard to move forward.

I've put it on GitHub. However the following is what I want reviewed.

This is app.py

from flask import Flask, url_for, request, json, Response, jsonify
from jsonschema import validate
import jsonschema
import sys
import json
import messages_handler
import message_schema as m_schema
import urllib
app = Flask(__name__)

mh = messages_handler.MessagesHandler()

# Entry point for requests
@app.route('/messages', methods = ['POST', 'GET'])
def api_message():

    if request.method == 'GET':
        return message_get()
    elif request.method == 'POST' and request.headers['Content-Type'] == 'application/json':
        return message_post(request.json)
        return handle_not_found()

# Handle Get requests for the API
def message_get():

    recipient, originator, message_contains = None, None, None

    if 'recipient' in request.args:
        recipient = int(request.args['recipient'])
    if 'originator' in request.args:
        originator = str(request.args['originator'])
    if 'message_contains' in request.args:
        message_contains = urllib.parse.unquote(str(request.args['message_contains'])).split()

    return jsonify(mh.search(recipient=recipient, originator=originator, content_contains=message_contains))

# Handles and Validates Post request for API
def message_post(post_json):

        validate(post_json, m_schema.MESSAGE_SCHEMA)
    except jsonschema.exceptions.ValidationError as ve:
        print("Record #{}: ERROR\n".format(post_json))
        print(str(ve) + "\n")
        return handleServerError(str(ve), 400)
    message = {
            'status': 200,
            'message': 'Message Inserted'
    resp = jsonify(message)
    return resp

# Helper function for handling 404
def handle_not_found(error=None):
    message = {
            'status': 404,
            'message': 'Not Found: ' + request.url,
    resp = jsonify(message)
    resp.status_code = 404

    return resp

# Helper function for showing errors
def handleServerError(error, code):
    message = {
            'status': code,
            'message': 'Error: ' + error,
    resp = jsonify(message)
    resp.status_code = code

    return resp

if __name__ == '__main__':


This is message_handler.py

import messages_storage
import messages_index
import message_schema as m_schema
import jsonschema
from jsonschema import validate
import datetime

# This handler class is responsible for the IO of stored and index messages
class MessagesHandler:

    def __init__(self):

        self.m_storage = messages_storage.MessagesStorage()
        self.m_index = messages_index.MessagesIndex()

        # index everything loaded from the file
        i = 0
        for message in self.m_storage.read_messages():
            self.m_index.index_message(message, i)
            i += 1

    # inserts message into system
    # returns false if failed
    def insert(self, message):

            validate(message, m_schema.MESSAGE_SCHEMA)
            message['timestamp'] = '{:%Y-%m-%d %H:%M:%S}'.format(datetime.datetime.now())
            m_id = self.m_storage.insert_message(message)
            if m_id is None: # If insert into storage failed
                print("Message Handler Insert Storage Error")
                return False
            self.m_index.index_message(message, m_id)
        except jsonschema.exceptions.ValidationError as ve:
            print("Message Handler Insert Error #{}: ERROR\n".format(message))
            print(str(ve) + "\n")
            return False
        return True

    # get messages with recipient
    # returns an array of messages json
    def search_recipients(self, recipient):

        m_ids = self.m_index.get_message_ids_recipient(str(recipient))
        return {"list":self.m_storage.read_messages(m_ids)}

    # get messages with originator
    # returns an array of messages json
    def search_originators(self, originator):

        m_ids = self.m_index.get_message_ids_originator(str(originator))
        return {"list":self.m_storage.read_messages(m_ids)}

    # get messages with message content containing words
    # returns an array of messages json
    def search_content(self, words):

        m_ids = self.m_index.get_message_ids_message_contains(words)
        return {"list":self.m_storage.read_messages(m_ids)}

    # get messages that fit all non None options
    def search(self, recipient = None, originator = None, content_contains = None):

        rec_set, ori_set, con_set = None, None, None
        intersection_list = []

        if recipient != None:
            rec_set = set(self.m_index.get_message_ids_recipient(str(recipient)))
        if originator != None:
            ori_set = set(self.m_index.get_message_ids_originator(str(originator)))
        if content_contains != None:
            con_set = set(self.m_index.get_message_ids_message_contains(content_contains))

        m_ids = list(set.intersection(*intersection_list))
        return {"list":self.m_storage.read_messages(m_ids)}

# Example use
if __name__ == '__main__':

    mh = MessagesHandler()

    message = {
        'message':'This is a test message.'

    print("Insert Message Status", mh.insert(message))
    print("Recipients search", mh.search_recipients(31612345678))
    print("Originator search", mh.search_originators('MessageBird'))
    print("Content search", mh.search_content(['This', 'is', 'a']))

This is message_index.py

import trie_tree

# This class is responsible for indexing
# and retriving from the index
# the indexed message properties
class MessagesIndex:
    def __init__(self):

        self.recipients_trie = trie_tree.Trie()
        self.originators_trie = trie_tree.Trie()
        self.messages_trie = trie_tree.Trie()

    # Indexes the message and all its components
    # Message assumed to be validated
    def index_message(self, message, message_id):

        self.index_recipient(message["recipient"], message_id)
        self.index_originator(message["originator"], message_id)
        self.index_content(message["message"], message_id)

    def index_recipient(self, recipient, message_id):

        self.recipients_trie.add(str(recipient), {"w":str(recipient),

    def index_originator(self, originator, message_id):

        self.originators_trie.add(originator, {"w":originator,

    def index_content(self, content, message_id):

        i = 0 # order of the word in message
        for word in content.split():
            self.messages_trie.add(word, {"w":word,
            i += 1

    # returns all message IDs by recipient
    # returns empty array if nothing found
    def get_message_ids_recipient(self, recipient):

        message_ids = []
        search_r = self.recipients_trie.search(str(recipient))
        if search_r != None:
            for data in search_r:
        return message_ids

    # returns all message IDs by originator
    # returns empty array if nothing found
    def get_message_ids_originator(self, originator):

        message_ids = []
        search_r = self.originators_trie.search(str(originator))
        if search_r != None:
            for data in search_r:
        return message_ids

    # returns all message IDs by text in message
    # returns empty array if nothing found
    def get_message_ids_message_contains(self, words):

        common_ids_set = set()
        i = 0 # words counted since first word is special
        for word in words:
            search_r = self.messages_trie.search(str(word))
            ids_set = set()
            if search_r != None:
                for data in search_r:
            else: # one word does not exist in the tree
                return []
            if i != 0:
                common_ids_set = common_ids_set.intersection(ids_set)
                # set is empty so it will remain empty. Exit with empty results
                if len(common_ids_set) == 0:
                    return []
            else: # First time just populate the set
                common_ids_set = ids_set
            i += 1
        return list(common_ids_set)

# Example use
if __name__ == '__main__':

    message = {
        'message':'This is a test message.'
    ms = MessagesIndex()
    ms.index_message(message, 0)

    print ("'31612345678' in recipients: ", ms.recipients_trie.contains('31612345678'))
    print ("'MessageBird' in originator: ", ms.originators_trie.contains('MessageBird'))
    print ("'This' in content: ", ms.messages_trie.contains('This'))
    print ("'is' in content: ", ms.messages_trie.contains('is'))

    print ("Search '31612345678' in recipients: ", ms.get_message_ids_recipient('31612345678'))
    print ("Search '316123452678232' in recipients: ", ms.get_message_ids_recipient('316123452678232'))
    print ("Search 'This is a' in message content: ", ms.get_message_ids_message_contains(['This', 'is', 'a']))
    print ("Search 'This is not a' in message content: ", ms.get_message_ids_message_contains(['This', 'is', 'a', 'not']))

This is message_storage.py

import mmap
import jsonschema
from jsonschema import validate
import io, json, os
import message_schema as schema
import array

class MessagesStorage:
    def __init__(self):

        # File storage for messages
        self.STORAGE_PATH = 'messages.txt'
        if not os.path.exists(self.STORAGE_PATH):
            open(self.STORAGE_PATH, 'w').close()

        # Key map storage for messages.txt
        self.MAP_PATH = 'map.txt'
        if not os.path.exists(self.MAP_PATH):
            open(self.MAP_PATH, 'w').close()

        # Holds all the messages inserted as message dictionary
        # ID to hold start and end position on file
        self._messages_pos = {}

        # Holds the number of messages
        self._id_counter = 0


    # inserts message json
    # returns insert ID uint
    # None if failed
    def insert_message(self, message):

        # Validates message json
            validate(message, schema.MESSAGE_SCHEMA)
        except jsonschema.exceptions.ValidationError as ve:
            print("Insert ERROR #{}: ERROR\n".format(post_json))
            print(str(ve) + "\n")
            return None

        with io.open(self.STORAGE_PATH, 'a', encoding='utf-8') as f:
            start_index = f.tell()
            f.write(json.dumps(message, ensure_ascii=False))
            end_index = f.tell()
            self._messages_pos[self._id_counter] = {"start":start_index, "end":end_index}
            self._id_counter += 1

        # Save keys in map.tx
        with io.open(self.MAP_PATH, 'a', encoding='utf-8') as f:
            #f.write(str(self._id_counter - 1)+' '+str(start_index)+' '+str(end_index)+';')
            print(str(self._id_counter - 1)+' '+str(start_index)+' '+str(end_index), file=f)

        return self._id_counter - 1

    # Reads message with id
    # returns array of json messages, all ids if no ids passed to function
    # returns empty array if nothing found
    def read_messages(self, message_ids=None):

            with open(self.STORAGE_PATH, "r+b") as f:
                mm = mmap.mmap(f.fileno(), 0)

                if message_ids != None:
                    return self.extract_messages(message_ids, mm)
                    all_ids = array.array('i',(0 for i in range(0,self._id_counter)))
                    return self.extract_messages(all_ids, mm)
        except ValueError:
            print ("No messages on storage")
            return []

    def extract_messages(self, message_ids, mm):

        result = []
        for message_id in message_ids:
            if message_id <= self._id_counter:
        return result

    # Initializes the object from file storage
    def init_from_file(self):

            with open(self.MAP_PATH,"r") as f:
                for line in f:
                        keys = line.split()
                        self._messages_pos[int(keys[0])] = {"start":int(keys[1]), "end":int(keys[2])}
                        self._id_counter += 1
                        print("Failed to inialize line", line)
            print("Failed to open keymap file")

# Example usage
if __name__ == '__main__':

    ms = MessagesStorage()

    message = {
        'message':'This is a test message.'


    print("Read message ", ms.read_messages([0]))
  • 2
    \$\begingroup\$ Huh, "Write everything from scratch: no frameworks" and from flask import Flask kinda contradict each other don't they? \$\endgroup\$ – Peilonrayz Sep 27 '17 at 9:21
  • \$\begingroup\$ Yes, I specifically asked if IO libs can be used and apparently I can. \$\endgroup\$ – Hassan Habbak Sep 27 '17 at 9:29
  • \$\begingroup\$ Is trie_tree of your own doing as well? \$\endgroup\$ – 301_Moved_Permanently Sep 27 '17 at 12:08
  • \$\begingroup\$ @MathiasEttinger Yes it is. :) \$\endgroup\$ – Peilonrayz Sep 27 '17 at 13:59

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.