I am currently working on building a small search engine (if it can be called that) that allows users to search for relevant articles from a short list of other older websites that do not have search features.

This particular part of the software takes a query from an API via a Redis Stream which is then used to sort a corpus according to the BM25 algorithm before returning the 30 most relevant results to an API via Redis.

While I have been using Python for small programming and web-development tasks for a few years, I am finally trying to improve the quality and performance of my code and as such I would appreciate it if more experienced developers than me could take a look at it and suggest how it may be improved.

In particular I would be interested if there is a better way to handle the Mongo database connections and pass them to the various classes that require access.

Thank you.

import os
import redis
import pymongo
from dotenv import load_dotenv
from rank_bm25 import BM25Plus
from redis.commands.json.path import Path as JPath

class Corpus():
    def __init__(self, mongo_col) -> None:
        self.col = mongo_col
        self.corpus = []

    def create_corpus(self) -> None:
        """ Creates Corpus Of Titles From A MongoDB Collection"""

        for data in self.col.find():
            self.corpus += data["title"]

        return self.corpus

class SearchEngine():
    def __init__(self, mongo_col_1, mongo_col_2, redis, corpus, stream_data) -> None:
        self.colA = mongo_col_1
        self.colB = mongo_col_2
        self.rdb = redis
        self.stream = stream_data
        self.corpus = corpus
        self.id = ""
        self.query = ""
        self.ranked_titles = []
        self.results = []

    def parse_stream(self) -> None:
        """ Parses Information From Redis Stream Message """

        s = self.stream[0][1][0][1]
        self.id = s["identifier"]
        self.query = s["query"]

    def search_engine(self) -> None:
        """Ranks Corpus According To Query Via BM25"""

        # Convert Query To Upper Case To Improve Search Results
        query = self.query.title()

        # BM25 Configuration
        tokenized_query = query.split(" ")
        tokenized_corpus = [title.split(" ") for title in self.corpus]
        bm25 = BM25Plus(tokenized_corpus)

        # Return "n" Most Relevant Titles
        self.ranked_titles = list(bm25.get_top_n(
            tokenized_query, self.corpus, n=30))

    def format_results(self) -> None:
        """ Formats Result's Title, URL & Source"""

        for title in self.ranked_titles:

            # Get Title's URL & Source
            data = self.colB.find_one({"title": title}, {
                "_id": 0, "url": 1, "source": 1})

            # Format As JSON
            result = {
                "title": title,
                "url": data["url"],
                "source": data["source"][0]

            self.results += [result]

    def send_results(self) -> None:
        """Sends JSON Formatted Results To API Via Redis"""

        data = {
            "id": self.id,
            "query": self.query,
            "results": self.results

        # Return Results To API Via REDIS DB1
        self.rdb.json().set(str("id:" + self.id), JPath.rootPath(), data)

        # Add Results To MongoDB Col1
        self.colA.insert_one({"_id": self.id, "data": [self.results_html]})

if __name__ == "__main__":
    # Load .env Variables From File

    # Redis .env Variables
    redis_host = os.getenv("REDIS_HOST")
    redis_port = os.getenv("REDIS_PORT")
    redis_password = os.getenv("REDIS_PASSWORD")

    # MongoDB General Settings
    mongo_port = os.getenv("MONGO_PORT")

    # MongoDB DB1 .env Variables
    mongo_host_1 = os.getenv("MONGO_HOST_1")
    mongo_db_1 = os.getenv("MONGO_DB_1")
    mongo_col_1 = os.getenv("MONGO_COL_1")

    # MongoDB DB2 .env Variables
    mongo_host_2 = os.getenv("MONGO_HOST_2")
    mongo_db_2 = os.getenv("MONGO_DB_2")
    mongo_col_2 = os.getenv("MONGO_COL_2")

    # Connect to Redis Streams
    # r0 -> Query Stream
    rdb0 = redis.Redis(host=redis_host, port=redis_port,
                       password=redis_password, db=0, decode_responses=True)

    # r1 -> Return Results To API
    rdb1 = redis.Redis(host=redis_host, port=redis_port,
                       password=redis_password, db=1, decode_responses=True)

    # Connect To MongoSE Database
    conn1 = pymongo.MongoClient(
    db1 = conn1[mongo_db_1]
    col1 = db1[mongo_col_1]

    # Connect To MongoCS Database
    conn2 = pymongo.MongoClient(
    db2 = conn2[mongo_db_2]
    col2 = db2[mongo_col_2]

    # Create Corpus
    c = Corpus(col2)
    c = c.create_corpus()

    # Block Stream To Wait For Incomming Message
    while True:
        stream = rdb0.xread({'streamA': "$"}, count=1, block=0)

        if stream != {}:
            s = SearchEngine(col1, col2, rdb1, c, stream)


Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Browse other questions tagged or ask your own question.