Skip to content

Commit

Permalink
Browse files Browse the repository at this point in the history
Connection with the database - handling multithreading petitions
  • Loading branch information
Javinator9889 committed Nov 5, 2019
1 parent 8b04a51 commit a8ea3d8
Show file tree
Hide file tree
Showing 7 changed files with 209 additions and 0 deletions.
5 changes: 5 additions & 0 deletions .gitignore
Expand Up @@ -105,3 +105,8 @@ venv.bak/

# keys folder
keys/

# audio files, if exists
*.m4a
*.mp3
*.ogg
6 changes: 6 additions & 0 deletions YouTubeMDBot/constants/__init__.py
Expand Up @@ -20,3 +20,9 @@
from ..constants.app_constants import PROGRAM_ARGS
from ..constants.app_constants import FFMPEG_OPENER
from ..constants.app_constants import FFMPEG_CONVERTER
from ..constants.app_constants import DB_HOST
from ..constants.app_constants import DB_NAME
from ..constants.app_constants import DB_PASSWORD
from ..constants.app_constants import DB_PORT
from ..constants.app_constants import DB_USER
from ..constants.app_constants import MAX_PROCESS
11 changes: 11 additions & 0 deletions YouTubeMDBot/constants/app_constants.py
Expand Up @@ -16,6 +16,8 @@
import os
import sys

from multiprocessing import cpu_count

PROGRAM_ARGS = sys.argv
# YouTube DL options
YDL_CLI_OPTIONS = ["youtube-dl", "--format", "bestaudio[ext=m4a]", "--quiet", "--output",
Expand All @@ -41,3 +43,12 @@
FFMPEG_OPENER = "ffmpeg -i - -f s16le -".split(" ")
FFMPEG_CONVERTER = ["ffmpeg", "-i", "-", "-vn", "-map_metadata", "0",
"-movflags", "use_metadata_tags"]

MAX_PROCESS = cpu_count()

# Database constants
DB_NAME = os.environ["DATABASE_NAME"]
DB_USER = os.environ["DATABASE_USER"]
DB_PASSWORD = os.environ["DATABASE_PASSWORD"]
DB_HOST = "127.0.0.1"
DB_PORT = 5432
17 changes: 17 additions & 0 deletions YouTubeMDBot/database/__init__.py
@@ -0,0 +1,17 @@
# YouTubeMDBot
# Copyright (C) 2019 - Javinator9889
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from ..database.psql import PostgresSQL
from ..database.query import Query
140 changes: 140 additions & 0 deletions YouTubeMDBot/database/psql.py
@@ -0,0 +1,140 @@
# YouTubeMDBot
# Copyright (C) 2019 - Javinator9889
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
from typing import Any
from typing import Optional
from queue import Queue
from psycopg2 import pool
from psycopg2.pool import PoolError
from threading import Condition
from threading import Lock
from multiprocessing import Process

from . import Query
from .. import MAX_PROCESS
from .. import DB_USER
from .. import DB_PORT
from .. import DB_PASSWORD
from .. import DB_NAME
from .. import DB_HOST


class PostgresSQL:
__instance = None

def __new__(cls, **kwargs):
if PostgresSQL.__instance is None:
PostgresSQL.__instance = object.__new__(cls)
PostgresSQL.__instance.__connection = \
pool.ThreadedConnectionPool(minconn=1,
maxconn=MAX_PROCESS,
user=DB_USER,
password=DB_PASSWORD,
dbname=DB_NAME,
host=DB_HOST,
port=DB_PORT)
PostgresSQL.__instance.__waiting_processes = Queue()
PostgresSQL.__instance.__running_processes = 0
PostgresSQL.__instance.__lock = Lock()
PostgresSQL.__instance.__queue_consumer = \
Process(target=cls.__consumer,
args=(cls, cls.__running_processes,))
PostgresSQL.__instance.__queue_consumer.start()
for key, value in kwargs.items():
setattr(PostgresSQL.__instance, key, value)
return PostgresSQL.__instance

def __consumer(self, process_queue: Queue):
condition = Condition()
while self.__connection.closed == 0:
with condition:
condition.wait_for(lambda: self.__running_processes <= MAX_PROCESS)
pending_process = process_queue.get()
pending_process["fn"](pending_process["query"])

def __get_connection(self) -> Optional[Any]:
with self.__lock:
if self.__running_processes <= MAX_PROCESS:
self.__running_processes += 1
try:
return self.__connection.getconn()
except PoolError:
return None
return None

def __free_connection(self, connection):
self.__connection.putconn(connection)
with self.__lock:
self.__running_processes -= 1

def execute(self, query: Query):
connection = self.__get_connection()
if connection is not None:
with connection.cursor() as cursor:
cursor.execute(query.query)
query._result = cursor.rowcount
query.is_completed = True
self.__free_connection(connection)
else:
self.__waiting_processes.put({"query": query, "fn": self.execute})

def fetchone(self, query: Query):
connection = self.__get_connection()
if connection is not None:
with connection.cursor() as cursor:
cursor.execute(query.query)
query._result = cursor.fetchone()
query.is_completed = True
self.__free_connection(connection)
else:
self.__waiting_processes.put({"query": query, "fn": self.fetchone})

def fetchall(self, query: Query):
connection = self.__get_connection()
if connection is not None:
with connection.cursor() as cursor:
cursor.execute(query.query)
query._result = cursor.fetchall()
query.is_completed = True
self.__free_connection(connection)
else:
self.__waiting_processes.put({"query": query, "fn": self.fetchall})

def fetchiter(self, query: Query):
connection = self.__get_connection()
if connection is not None:
with connection.cursor() as cursor:
cursor.execute(query.query)

def generate():
while True:
items = cursor.fetchmany()
if not items:
break
for item in items:
yield item

query._result = generate()
query.is_completed = True
self.__free_connection(connection)
else:
self.__waiting_processes.put({"query": query, "fn": self.fetchiter})

def __del__(self):
if self.__connection.closed == 0:
while not self.__waiting_processes.empty():
continue
self.__queue_consumer.terminate()
self.__instance.__connection.closeall()
29 changes: 29 additions & 0 deletions YouTubeMDBot/database/query.py
@@ -0,0 +1,29 @@
# YouTubeMDBot
# Copyright (C) 2019 - Javinator9889
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import threading


class Query:
def __init__(self, query: str):
self.query = query
self.is_completed = False
self._result = None
self._condition = threading.Condition()

def result(self):
with self._condition:
self._condition.wait_for(lambda: self.is_completed)
return self._result
1 change: 1 addition & 0 deletions YouTubeMDBot/requirements.txt
Expand Up @@ -6,3 +6,4 @@ ujson
youtube_dl
pyacoustid
python-telegram-bot
psycopg2

0 comments on commit a8ea3d8

Please sign in to comment.