Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# -*- coding: utf-8 -*-
import logging
import os
import sqlite3
from .abstractdb import AbstractDB
class SQLiteDB(AbstractDB):
def __init__(self, dbpath="pastepwn"):
super().__init__()
self.dbpath = dbpath
self.logger = logging.getLogger(__name__)
self.logger.debug("Initializing SQLite - {0}".format(dbpath))
# Check if the folder path exists
if not os.path.exists(os.path.dirname(dbpath)):
# If not, create the path and the file
os.mkdir(os.path.dirname(dbpath))
open(self.dbpath, "a").close()
try:
self.db = sqlite3.connect(dbpath, check_same_thread=False)
self.db.text_factory = lambda x: str(x, 'utf-8', "ignore")
# -*- coding: utf-8 -*-
import logging
import mysql.connector
from .abstractdb import AbstractDB
class MysqlDB(AbstractDB):
def __init__(self, ip="127.0.0.1", port=3306, unix_socket=None, dbname="pastepwn", username=None, password=None, timeout=10):
super().__init__()
self.logger = logging.getLogger(__name__)
self.logger.debug("Initializing MySQLDB - {0}:{1}".format(ip, port))
if unix_socket:
self.db = mysql.connector.connect(
host=ip,
user=username,
passwd=password,
unix_socket=unix_socket,
connection_timeout=timeout
)
else:
self.db = mysql.connector.connect(
# -*- coding: utf-8 -*-
import logging
import pymongo
from pymongo.errors import ConnectionFailure
from .abstractdb import AbstractDB
class MongoDB(AbstractDB):
def __init__(self, ip="127.0.0.1", port=27017, dbname="pastepwn", collectionname="pastes"):
super().__init__()
self.logger = logging.getLogger(__name__)
self.logger.debug("Initializing MongoDB - {0}:{1}".format(ip, port))
self.db = pymongo.MongoClient(ip, port, serverSelectionTimeoutMS=5000)
try:
self.db.admin.command("ismaster")
except ConnectionFailure as e:
self.logger.error(e)
raise e
self.logger.debug("Connected to database!")
self.db = self.db[dbname]