57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
import logging
|
|
|
|
from cloudant.client import CouchDB
|
|
|
|
|
|
class DB:
|
|
def __init__(self, config: dict) -> None:
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
self.username = config['username']
|
|
self.password = config['password']
|
|
self.url = config['url']
|
|
self.client = None
|
|
self.session = None
|
|
|
|
def initialize_shortenit(self):
|
|
try:
|
|
self.counter_db = self.client['counter']
|
|
except KeyError:
|
|
self.logger.warn(
|
|
"The 'counter' database was not found, creating...")
|
|
self.counter_db = self.client.create_database('counter')
|
|
if self.counter_db.exists():
|
|
self.logger.info(
|
|
"The 'counter' database was successfully created.")
|
|
|
|
try:
|
|
self.data_db = self.client['data']
|
|
except KeyError:
|
|
self.logger.warn(
|
|
"The 'data' database was not found, creating...")
|
|
self.data_db = self.client.create_database('data')
|
|
if self.data_db.exists():
|
|
self.logger.info(
|
|
"The 'data' database was successfully created.")
|
|
|
|
try:
|
|
self.pointers_db = self.client['pointers']
|
|
except KeyError:
|
|
self.logger.warn(
|
|
"The 'pointers' database was not found, creating...")
|
|
self.pointers_db = self.client.create_database('pointers')
|
|
if self.pointers_db.exists():
|
|
self.logger.info(
|
|
"The 'pointers' database was successfully created.")
|
|
|
|
def __enter__(self) -> CouchDB:
|
|
"""
|
|
"""
|
|
self.client = CouchDB(self.username, self.password,
|
|
url=self.url, connect=True)
|
|
self.session = self.client.session()
|
|
return self
|
|
|
|
def __exit__(self, *args) -> None:
|
|
"""
|
|
"""
|
|
self.client.disconnect()
|