Converted to SQLAlchemy

Converted from straight Postgres to SQLAlchemy.

.gitignore

@@ -1,6 +1,4 @@
-
 .idea/
-
 __pycache__/
-
 venv/
+*.db

config.toml

@@ -0,0 +1,12 @@
+name = "DNS Server Database Configuration"
+[database]
+database_type = "postgresql"
+database_dbapi = 'psycopg2' # Options [default, psycopg2, pg8000] Recommend psycopg2
+database_host = '127.0.0.1:5433'
+database_name = 'fakeDNS'
+database_user = 'postgres'
+database_password = '1qaz2wsx!QAZ@WSX'
+
+[dnsServer]
+id = 'ad54cb25-85a1-4809-b144-16e50fd251bb'
+hostname = 'dns.google.com'

database/__init__.py

@@ -0,0 +1 @@
+from .database import Database

database/database.py

@@ -0,0 +1,113 @@
+import os
+
+import sqlalchemy.databases
+import sqlalchemy.exc
+import toml
+from sqlalchemy import *
+from sqlalchemy.dialects import postgresql
+from sqlalchemy.dialects import mysql
+from sqlalchemy.dialects import mssql
+from sqlalchemy.dialects import oracle
+from sqlalchemy.orm import sessionmaker
+from .sql_alchemy_base import Base
+from urllib.parse import quote
+
+class Database:
+    metadata = MetaData()
+    Session: sessionmaker
+    engine: sqlalchemy.engine
+    bool_connected = False
+    bool_sqlite = False
+    database_config: dict
+    def __init__(self, database_config: dict):
+        try:
+            self.database_config = database_config
+            database_url = self.create_database_url()
+            if not database_url['bool_error']:
+                # print(database_url)
+                conn = self.connect(database_url['database_url'])
+                # print(conn)
+                if conn['bool_connected']:
+                    self.bool_connected = True
+                Base.metadata.create_all(self.engine)
+                self.Session = sessionmaker(bind=self.engine)
+            else:
+                print(database_url['str_error'])
+        except Exception as e:
+            print(e)
+
+    def create_database_url(self):
+        dict_return = {'bool_error': False, 'str_error': '', 'database_url': None}
+        try:
+            database_configuration = self.database_config
+            if database_configuration['database_type'] == 'postgresql':
+                passwd = quote(database_configuration['database_password'])
+                db_api = ""
+                if database_configuration['database_dbapi'] == "psycopg2" or\
+                        database_configuration['database_dbapi'] == "pg8000":
+                    db_api = "+{}".format(database_configuration['database_dbapi'])
+                database_url = 'postgresql{}://{}:{}@{}/{}'.format(db_api,database_configuration['database_user'],
+                                                                   passwd,
+                                                                   database_configuration['database_host'],
+                                                                   database_configuration['database_name'])
+                dict_return['database_url'] = database_url
+        except Exception as e:
+            dict_return['bool_error'] = True
+            dict_return['str_error'] = str(e)
+        return dict_return
+
+    def connect(self, database_url):
+        dict_return = {'bool_error': False, 'str_error': '', 'bool_connected': False}
+        try:
+            self.engine = create_engine(database_url)
+            if self.engine.connect():
+                dict_return['bool_connected'] = True
+        except sqlalchemy.exc.OperationalError as e:
+            dict_return['bool_error'] = False
+            dict_return['str_error'] = str(e)
+        except Exception as e:
+            dict_return['bool_error'] = False
+            dict_return['str_error'] = str(e)
+        return dict_return
+
+    class IPv4_Networks(Base):
+        __tablename__ = 'IPv4_Networks'
+        __table_args__ = {'extend_existing': True}
+        network = Column(String(20), primary_key=True)
+        country_id = Column(Integer(), nullable=False)
+
+    class IPv6_Networks(Base):
+        __tablename__ = 'IPv6_Networks'
+        __table_args__ = {'extend_existing': True}
+        network = Column(String(43), primary_key=True)
+        country_id = Column(Integer(), nullable=False)
+
+    class Country_List(Base):
+        __tablename__ = 'country_list'
+        __table_args__ = {'extend_existing': True}
+        country_id = Column(Integer(),primary_key=True, nullable=False)
+        country_code = Column(String(5), nullable=False)
+        country_name = Column(String(255), nullable=False)
+        ccTLD = Column(String(4), nullable=True)
+
+    class Dns_Authoritative(Base):
+        __tablename__ = 'dns_authoritative'
+        __table_args__ = {'extend_existing': True}
+        entry_text = Column(String(255),primary_key=True, nullable=False)
+        bool_authoritative = Column(Boolean(), nullable=False)
+        authoritative_data = Column(Text(), nullable=True)
+
+    class Dns_Entries(Base):
+        __tablename__ = 'dns_entries'
+        __table_args__ = {'extend_existing': True}
+        id = Column(Integer(), primary_key=True,autoincrement = True)
+        entry_text = Column(String(255),primary_key=False, nullable=False)
+        record_type = Column(Integer(), nullable=False)
+        record_data = Column(JSON(), nullable=False)
+
+    class Tld_Listing(Base):
+        __tablename__ = 'tld_listing'
+        __table_args__ = {'extend_existing': True}
+        domain = Column(String(20),primary_key=True, nullable=False)
+        type = Column(String(20), nullable=False)
+        org = Column(String(255), nullable=True)

database/sql_alchemy_base.py

@@ -0,0 +1,3 @@
+from sqlalchemy.ext.declarative import declarative_base
+Base = declarative_base()
+

database_config_example.toml

@@ -0,0 +1,37 @@
+name = "DNS Server Database Configuration Example"
+# Only use one of the following in the database_config.tml file
+[database]
+database_type = "sqlite"
+database_path = '\full\path\to\database\file.db' # File will be created
+
+[database]
+database_type = "postgresql"
+database_dbapi = 'psycopg2' # Options [default, psycopg2, pg8000] Recommend psycopg2
+database_host = '127.0.0.1'
+database_name = 'myDatabase'
+database_user = 'admin'
+database_password = 'P@55w0rd!'
+
+[database]
+database_type = "mysql" # mysql and mariadb
+database_dbapi = 'mysqldb' # Options [default, mysqldb, pymysql] Recommend mysqldb
+database_host = '127.0.0.1'
+database_name = 'myDatabase'
+database_user = 'admin'
+database_password = 'P@55w0rd!'
+
+[database]
+database_type = "oracle" # Oracle Database
+database_dbapi = 'default' # Options [default, cx_oracle] Recommend default
+database_host = '127.0.0.1'
+database_name = 'myDatabase'
+database_user = 'admin'
+database_password = 'P@55w0rd!'
+
+[database]
+database_type = "mssql" # Microsoft SQL Server
+database_dbapi = 'pyodbc' # Options [pyodbc, pymssql] Recommend pyodbc
+database_host = '127.0.0.1'
+database_name = 'myDatabase'
+database_user = 'admin'
+database_password = 'P@55w0rd!'