Merge branch 'feature/auto_with_checkresults' into feature/automatic

db/local_db_manager.py

@@ -0,0 +1,63 @@
+from builtins import list
+
+import sqlalchemy
+from sqlalchemy import MetaData, Column, String, Integer, DateTime, Table
+from sqlalchemy.orm import Session
+from sqlalchemy_utils import database_exists, create_database
+
+import definitions
+import params
+from pojo.ReserveResultPojo import ReserveResultPojo
+from pojo.captcha_error_contact_pojo import ContactInErrorPojo, ERROR_TYPE_CAPTCHA
+
+
+class LocalDbManager:
+
+    def __init__(self, path: str):
+        self.session = Session(self.init_db(path))
+
+    def init_db(self, path: str):
+        uri_for_db = "sqlite:///{}/{}.db?check_same_thread=false".format(path, "appointment")
+        print(uri_for_db)
+        # 2.-Turn on database engine
+        db_engine = sqlalchemy.create_engine(uri_for_db)  # ensure this is the correct path for the sqlite file.
+        if not database_exists(uri_for_db):
+            create_database(uri_for_db)
+        connextion = db_engine.connect()
+        if not db_engine.dialect.has_table(connextion,
+                                           ContactInErrorPojo.__tablename__):  # If table don't exist, Create.
+            metadata = MetaData(db_engine)
+            # Create a table with the appropriate Columns
+            Table(ContactInErrorPojo.__tablename__, metadata,
+                  Column('id', Integer, primary_key=True, autoincrement=True),
+                  Column('mail', String),
+                  Column('phone', String),
+                  Column('passport', String),
+                  Column('last_name', String),
+                  Column('first_name', String),
+                  Column('ccid', String),
+                  Column('position', Integer),
+                  Column('error_type', Integer),
+                  Column('update_at', DateTime),
+                  Column('create_at', DateTime))
+            # Implement the creation
+            metadata.create_all()
+
+        return db_engine
+
+    def insert_or_update(self, instance: ContactInErrorPojo):
+        self.session.merge(instance)
+        self.session.commit()
+
+    def get_all_captcha_error_contacts(self) -> list:
+        return self.session.query(ContactInErrorPojo).filter(ContactInErrorPojo.error_type == ERROR_TYPE_CAPTCHA).all()
+
+    def handle_success(self, reservePojo: ReserveResultPojo):
+        #         delete the contact from table
+        self.session.query(ContactInErrorPojo).filter(ContactInErrorPojo.mail == reservePojo.email).delete()
+
+
+if __name__ == '__main__':
+    list = params.local_db_manager.get_all_captcha_error_contacts()
+    for item in list:
+        print(item.mail)