removed position repo

switched position to static enum
add export to user repo

src/db/Position.py

@@ -1,5 +1,32 @@
-from pydantic import BaseModel
+from enum import Enum
 
 
-class Position(BaseModel):
+class Position(Enum):
-    name: str
+    PRODUCT_MANAGEMENT = "Product Management"
+    DATA_ANALYST = "Data Analyst"
+    DATA_SCIENCE = "Data Science, ML & Algorithms"
+    SOFTWARE_ENGINEERING = "Software Engineering"
+    FULLSTACK_DEVELOPMENT = "Fullstack Development"
+    QA = "QA"
+    CYBERSECURITY = "Cybersecurity"
+    IT_AND_SYSTEM_ADMINISTRATION = "IT and System Administration"
+    FRONTEND_DEVELOPMENT = "Frontend Development"
+    DEV_OPS = "DevOps"
+    UI_UX = "UI/UX, Design & Content"
+    HR_RECRUITMENT = "HR & Recruitment"
+    MOBILE_DEVELOPMENT = "Mobile Development"
+    HARDWARE_ENGINEERING = "Hardware Engineering"
+    EMBEDDED_ENGINEERING = "Embedded, Low Level & Firmware Engineering",
+    CUSTOMER_SUCCESS = "Customer Success"
+    PROJECT_MANAGEMENT = "Project Management"
+    OPERATIONS = "Operations"
+    FINANCE = "Finance"
+    SYSTEMS_ENGINEERING = "Systems Engineering"
+    MARKETING = "Marketing"
+    SALES = "Sales"
+    LEGAL_POLICY = "Compliance, Legal & Policy"
+    C_LEVEL = "C-Level"
+    BUSINESS_DEVELOPMENT = "Business Development"
+    MECHANICAL_ENGINEERING = "Mechanical Engineering"
+    NATURAL_SCIENCE = "Natural Science"
+    OTHER = "Other"

src/db/User.py

@@ -1,7 +1,6 @@
-from datetime import datetime
 from typing import Optional, Union
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
 
 from db.Position import Position
 
@@ -9,5 +8,5 @@ from db.Position import Position
 class User(BaseModel):
     full_name: str
     username: str
-    chat_id: Union[int,str]
+    chat_id: Union[int, str] = None
-    field: Optional[Position] = None
+    field: Optional[Position] = Position.SOFTWARE_ENGINEERING

src/db/position_repository.py

@@ -1,106 +0,0 @@
-from typing import Optional
-
-from dotenv import load_dotenv
-from pymongo import UpdateOne
-
-from jobspy import create_logger
-from .Position import Position
-from .monogo_db import mongo_client
-
-load_dotenv()
-
-
-class PositionRepository:
-    _instance = None
-
-    def __new__(cls):
-
-        if cls._instance is not None:
-            return cls._instance
-
-        self = super().__new__(cls)
-        cls._instance = self
-        self.logger = create_logger("PositionRepository")
-        self.collection = mongo_client.db["field"]
-        return cls._instance
-
-    def find_all(self) -> list[Position]:
-        positions = list(self.collection.find({}))
-        return [Position(**position) for position in positions]
-
-    def find_by_id(self, position_id: str) -> Optional[Position]:
-        """
-        Finds a position document in the collection by its ID.
-
-        Args:
-            position_id: The ID of the position to find.
-
-        Returns:
-            The position document if found, otherwise None.
-        """
-        result = self.collection.find_one({"id": position_id})
-        return Position(**result)
-
-    def update(self, position: Position) -> bool:
-        """
-        Updates a Position in the database.
-
-        Args:
-            position: A dictionary representing the Position data.
-
-        Returns:
-            True if the update was successful, False otherwise.
-        """
-        result = self.collection.update_one({"id": position.id}, {"$set": position.model_dump()})
-        return result.modified_count > 0
-
-    def insert_position(self, position: Position):
-        """
-        Inserts a new position posting into the database collection.
-
-        Args:
-            position (Position): The Position object to be inserted.
-
-        Raises:
-            Exception: If an error occurs during insertion.
-        """
-        self.collection.insert_one(position.model_dump())
-        self.logger.info(f"Inserted new position with name {position.name}.")
-
-    def insert_many_if_not_found(self, positions: list[Position]) -> tuple[list[Position], list[Position]]:
-        """
-        Perform bulk upserts for a list of Position objects into a MongoDB collection.
-        Only insert new positions and return the list of newly inserted positions.
-        """
-        operations = []
-        new_positions = []  # List to store the new positions inserted into MongoDB
-        old_positions = []  # List to store the new positions inserted into MongoDB
-        for position in positions:
-            position_dict = position.model_dump()
-            operations.append(
-                UpdateOne(
-                    {"id": position.id},  # Match by `id`
-                    # Only set positions if the position is being inserted (not updated)
-                    {"$setOnInsert": position_dict},
-                    upsert=True  # Insert if not found, but do not update if already exists
-                )
-            )
-
-        if operations:
-            # Execute all operations in bulk
-            result = self.collection.bulk_write(operations)
-            self.logger.info(f"Matched: {result.matched_count}, Upserts: {
-            result.upserted_count}, Modified: {result.modified_count}")
-
-            # Get the newly inserted positions (those that were upserted)
-            # The `upserted_count` corresponds to how many new documents were inserted
-            for i, position in enumerate(positions):
-                if result.upserted_count > 0 and i < result.upserted_count:
-                    new_positions.append(position)
-                else:
-                    old_positions.append(position)
-
-        return old_positions, new_positions
-
-
-position_repository = PositionRepository()

src/db/user_repository.py

@@ -22,6 +22,7 @@ class UserRepository:
         cls._instance = self
         self.logger = create_logger("UserRepository")
         self.collection = mongo_client.db["user"]
+        self.collection.create_index('username', unique=True)
         return cls._instance
 
     def find_by_id(self, user_id: str) -> Optional[User]:
@@ -97,3 +98,5 @@ class UserRepository:
                     old_users.append(user)
 
         return old_users, new_users
+
+user_repository = UserRepository()

src/telegram_handler/telegram_start_handler.py

@@ -5,9 +5,9 @@ from telegram.ext import (
     ContextTypes, ConversationHandler, CommandHandler, MessageHandler, filters,
 )
 
+from db.Position import Position
 from db.User import User
-from db.position_repository import position_repository
+from db.user_repository import UserRepository, user_repository
-from db.user_repository import UserRepository
 from jobspy.scrapers.utils import create_logger
 from telegram_bot import TelegramBot
 from telegram_handler.start_handler_constats import START_MESSAGE, POSITION_MESSAGE, POSITION_NOT_FOUND, \
@@ -30,9 +30,7 @@ class TelegramStartHandler(TelegramHandler):
     def __init__(self):
         self.filters = None
         self.telegram_bot = TelegramBot()
-        self.user_repository = UserRepository()
         self.logger = create_logger("TelegramStartHandler")
-        self.positions = position_repository.find_all()
         self.temp_user = None
         self.cities = None
 
@@ -40,9 +38,9 @@ class TelegramStartHandler(TelegramHandler):
         """Starts the conversation and asks the user about their position."""
         chat: Chat = update.message.chat
         user = User(full_name=chat.full_name, username=chat.username, chat_id=chat.id)
-        self.user_repository.insert_user(user)
+        user_repository.insert_user(user)
 
-        buttons = [[KeyboardButton(position.name)] for position in self.positions]
+        buttons = [[KeyboardButton(position.value)] for position in Position]
         reply_markup = ReplyKeyboardMarkup(buttons, one_time_keyboard=True,
                                            input_field_placeholder=Flow.POSITION.name)
         await update.message.reply_text(
@@ -147,7 +145,7 @@ class TelegramStartHandler(TelegramHandler):
         # chat.username - 'Qw1zeR'
         # chat.full_name - 'Qw1zeR'
         # user = User(full_name=chat.full_name, username=chat.username, chat_id=chat.id)
-        # self.user_repository.insert_user(user)
+        # user_repository.insert_user(user)
         # fields = field_repository.find_all()  # Get all fields from the database
         # buttons = [[KeyboardButton(field.name)] for field in fields]
         # reply_markup = ReplyKeyboardMarkup(buttons, one_time_keyboard=True)