Python Script that searches a corpus for most relevant results according to search terms

Question

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
        # ONLY USED BY METRIX SERVICE
        self.colA.insert_one({"_id": self.id, "data": [self.results_html]})


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

    # 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(
        host=f"mongodb://{mongo_host_1}:{str(mongo_port)}/")
    db1 = conn1[mongo_db_1]
    col1 = db1[mongo_col_1]

    # Connect To MongoCS Database
    conn2 = pymongo.MongoClient(
        host=f"mongodb://{mongo_host_2}:{str(mongo_port)}/")
    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)
            s.parse_stream()
            s.search_engine()
            s.format_results()
            s.send_results()