check_usernames_v3

tasks/check_usernames/core/connection/base_connection.py

@@ -0,0 +1,38 @@
+from abc import ABC, abstractmethod
+
+
+class BaseConnection(ABC):
+    """
+    Abstract base class for connection operations with databases and APIs.
+    This interface provides methods for connecting, disconnecting, and checking the connection status.
+
+    Implementations of this interface can be used with MySQL, SQLite, or other databases,
+    as well as with API setups.
+    """
+
+    @abstractmethod
+    def connect(self):
+        """
+        Connects to the database or API.
+
+        Raises:
+            ConnectionError: If the connection fails.
+        """
+        pass
+
+    @abstractmethod
+    def disconnect(self):
+        """
+        Disconnects from the database or API.
+        """
+        pass
+
+    @abstractmethod
+    def check(self):
+        """
+        Checks the connection status.
+
+        Returns:
+            bool: True if the connection is active, False otherwise.
+        """
+        pass

tasks/check_usernames/core/connection/mysql_connection.py

@@ -0,0 +1,78 @@
+from __future__ import annotations
+
+import pymysql
+
+from tasks.check_usernames.core.connection.base_connection import BaseConnection
+
+
+class MySQLBaseConnection(BaseConnection):
+    """
+    Implementation of the BaseConnection interface for MySQL databases using the pymysql library.
+    """
+
+    def __init__(self, host: str, port: str | int, database: str, db_connect_file: str = None, user: str = None,
+                 password: str = None):
+        """
+        Initializes a MySQLBaseConnection object with the provided connection parameters.
+
+        Parameters:
+            host (str): The host name or IP address of the MySQL server.
+            port (int): The port number of the MySQL server.
+            database (str): The name of the MySQL database to connect to.
+            db_connect_file (str, optional): The path to the MySQL database connection file. Defaults to None.
+            user (str, optional): The name of the MySQL user. Defaults to None.
+            password (str, optional): The password of the MySQL user. Defaults to None.
+        """
+
+        self.host = host
+        self.port = port
+        self.user = user
+        self.password = password
+        self.database = database
+        self.connection = None
+        self.db_connect_file = db_connect_file
+
+    def connect(self):
+        """
+        Connects to the MySQL database using the provided parameters.
+
+        Raises:
+            pymysql.err.OperationalError: If the connection fails.
+        """
+        if self.db_connect_file is not None:
+            self.connection = pymysql.connect(
+                host=self.host,
+                read_default_file=self.db_connect_file,
+                db=self.database,
+                charset='utf8mb4',
+                port=self.port,
+                cursorclass=pymysql.cursors.DictCursor,
+            )
+        else:
+            self.connection = pymysql.connect(
+                host=self.host,
+                port=self.port,
+                charset='utf8mb4',
+                user=self.user,
+                password=self.password,
+                db=self.database
+            )
+
+    def disconnect(self):
+        """
+        Disconnects from the MySQL database.
+        """
+        if self.connection:
+            self.connection.close()
+            self.connection = None
+
+    def check(self):
+        """
+        Checks the status of the MySQL database connection.
+
+        Returns:
+            bool: True if the connection is open, False otherwise.
+        """
+        if self.connection:
+            return self.connection.open
+        return False

tasks/check_usernames/core/connection/sqlite_connection.py

@@ -0,0 +1,47 @@
+import sqlite3
+
+from tasks.check_usernames.core.connection.base_connection import BaseConnection
+
+
+class SQLiteBaseConnection(BaseConnection):
+    """
+    Implementation of the BaseConnection interface for SQLite databases using the sqlite3 library.
+    """
+
+    def __init__(self, database_file: str):
+        """
+        Initializes a SQLiteBaseConnection object with the path to the SQLite database file.
+
+        Parameters:
+            database_file (str): The path to the SQLite database file.
+        """
+        self.database_file = database_file
+        self.connection = None
+
+    def connect(self):
+        """
+        Connects to the SQLite database using the provided database file path.
+
+        Raises:
+            sqlite3.Error: If the connection fails.
+        """
+        self.connection = sqlite3.connect(self.database_file)
+
+    def disconnect(self):
+        """
+        Disconnects from the SQLite database.
+        """
+        if self.connection:
+            self.connection.close()
+            self.connection = None
+
+    def check(self):
+        """
+        Checks the status of the SQLite database connection.
+
+        Returns:
+            bool: True if the connection is open, False otherwise.
+        """
+        if self.connection:
+            return True
+        return False

tasks/check_usernames/core/factory/factory.py

@@ -0,0 +1,36 @@
+from pywikibot import config as _config
+
+from tasks.check_usernames.core.connection.mysql_connection import MySQLBaseConnection
+from tasks.check_usernames.core.connection.sqlite_connection import SQLiteBaseConnection
+from tasks.check_usernames.core.persistence.mysql_persistence import MySQLPersistence
+from tasks.check_usernames.core.persistence.sqlite_persistence import SQLitePersistence
+from tasks.check_usernames.core.repositories.mysql_repository import MySQLRepository
+from tasks.check_usernames.core.repositories.sqlite_repository import SQLiteRepository
+
+
+class DatabaseFactory:
+    def create_mysql_repository(self) -> MySQLRepository:
+        connection = MySQLBaseConnection(
+            host=_config.db_hostname_format.format("arwiki"),
+            port=_config.db_port,
+            database=_config.db_name_format.format("arwiki"),
+            db_connect_file=_config.db_connect_file,
+        )
+        connection.connect()
+        return MySQLRepository(
+            persistence=MySQLPersistence(
+                connection=connection
+            )
+        )
+
+    def create_sqlite_memory_repository(self) -> SQLiteRepository:
+        connection = SQLiteBaseConnection(
+            # database_file="file::memory:?cache=shared",
+            database_file="/home/lokas/PycharmProjects/pythonProject3/code/tasks/check_usernames/core/demo.db",
+        )
+        connection.connect()
+        return SQLiteRepository(
+            persistence=SQLitePersistence(
+                connection=connection
+            )
+        )

tasks/check_usernames/core/models/user.py

@@ -0,0 +1,14 @@
+class UserStatus:
+    STATE_GET_FROM_DB = 0
+    STATE_SEND_TO_API = 1
+    STATE_ACCEPT_ON_API = 2
+    STATE_REJECT_ON_API = 3
+
+
+class User:
+
+    def __init__(self, id, user_name, created_at, status=UserStatus.STATE_GET_FROM_DB):
+        self.id = id
+        self.user_name = user_name
+        self.created_at = created_at
+        self.status = status

tasks/check_usernames/core/persistence/base_persistence.py

@@ -0,0 +1,70 @@
+from abc import ABC, abstractmethod
+
+
+class BasePersistence(ABC):
+
+    @abstractmethod
+    def execute(self, query: str, params=None):
+        """
+        Executes a custom SQL query.
+
+        :param query: The SQL query to execute.
+
+        :type query: str
+
+        :param params: Optional parameters to be substituted in the query.
+
+        :type params: Union[None, tuple]
+
+        """
+        pass
+
+    @abstractmethod
+    def select(self, query: str, params=None):
+        """
+        Executes a SELECT query and returns the result as a list of rows.
+
+        Parameters:
+            query (str): The SQL query string to execute.
+            params (tuple or dict, optional): Parameters for the query (default: None).
+
+        Returns:
+            list: A list of rows returned by the query.
+        """
+        pass
+
+    @abstractmethod
+    def select_one(self, query: str, params=None):
+        """
+        Executes a SELECT query and returns a single row.
+
+        Parameters:
+            query (str): The SQL query string to execute.
+            params (tuple or dict, optional): Parameters for the query (default: None).
+
+        Returns:
+            tuple or None: A single row returned by the query or None if no rows are found.
+        """
+        pass
+
+    @abstractmethod
+    def delete(self, query: str, params=None):
+        """
+        Executes a DELETE query.
+
+        Parameters:
+            query (str): The SQL query string to execute.
+            params (tuple or dict, optional): Parameters for the query (default: None).
+        """
+        pass
+
+    @abstractmethod
+    def update(self, query: str, params=None):
+        """
+        Executes an UPDATE query.
+
+        Parameters:
+            query (str): The SQL query string to execute.
+            params (tuple or dict, optional): Parameters for the query (default: None).
+        """
+        pass

tasks/check_usernames/core/persistence/mysql_persistence.py

@@ -0,0 +1,87 @@
+from tasks.check_usernames.core.connection.base_connection import BaseConnection
+from tasks.check_usernames.core.persistence.base_persistence import BasePersistence
+
+
+class MySQLPersistence(BasePersistence):
+    def __init__(self, connection: BaseConnection):
+        """
+        Initializes a new instance of the MySQLPersistence class.
+
+        :param connection: The connection object to use for database operations.
+        :type connection: BaseConnection
+        """
+        self.connection = connection
+
+    def delete(self, query, params=None):
+        """
+        Deletes data from the database based on the given query and parameters.
+
+        :param query: The SQL query used to delete data from the database.
+        :type query: str
+        :param params: Optional parameters to be substituted in the query.
+        :type params: Union[None, tuple]
+        """
+        with self.connection.connection.cursor() as cursor:
+            cursor.execute(query, params)
+            self.connection.connection.commit()
+
+    def execute(self, query, params=None):
+        """
+        Executes a custom SQL query.
+
+        :param query: The SQL query to execute.
+        :type query: str
+        :param params: Optional parameters to be substituted in the query.
+        :type params: Union[None, tuple]
+        """
+        with self.connection.connection.cursor() as cursor:
+            cursor.execute(query, params)
+            self.connection.connection.commit()
+
+    def update(self, query, params=None):
+        """
+        Updates data in the database based on the given query and parameters.
+
+        :param query: The SQL query used to update data in the database.
+        :type query: str
+        :param params: Optional parameters to be substituted in the query.
+        :type params: Union[None, tuple]
+        """
+        with self.connection.connection.cursor() as cursor:
+            cursor.execute(query, params)
+            self.connection.connection.commit()
+
+    def select(self, query, params=None):
+        """
+        Executes a SELECT query and returns the results as a list of rows.
+
+        :param query: The SQL query to execute.
+        :type query: str
+        :param params: Optional parameters to be substituted in the query.
+        :type params: Union[None, tuple]
+
+        :return: A list of rows returned by the query.
+        :rtype: list
+        """
+        with self.connection.connection.cursor() as cursor:
+            print(query, params)
+            cursor.execute(query, params)
+            result = cursor.fetchall()
+        return result
+
+    def select_one(self, query, params=None):
+        """
+        Executes a SELECT query and returns a single row.
+
+        :param query: The SQL query to execute.
+        :type query: str
+        :param params: Optional parameters to be substituted in the query.
+        :type params: Union[None, tuple]
+
+        :return: A single row returned by the query, or None if no rows are found.
+        :rtype: Union[tuple, None]
+        """
+        with self.connection.connection.cursor() as cursor:
+            cursor.execute(query, params)
+            result = cursor.fetchone()
+        return result