How to use __close_db method in SeleniumBase

Best Python code snippet using SeleniumBase

Db.py

Source:Db.py

...5from models.Book import Book6from models.Lend import Lend7def __open_db():  # Private method untuk buka/tutup koneksi database8    return sqlite3.connect(application.config['DB_NAME'])9def __close_db(cursor, con):10    cursor.close()11    con.close()12def init_db():  # Inisialisasi database (jika tidak ada table, create new table)13    con = __open_db()14    cursor = con.cursor()15    query = 'CREATE TABLE IF NOT EXISTS user (' \16            'nim TEXT PRIMARY KEY NOT NULL,' \17            'name TEXT NOT NULL,' \18            'password TEXT NOT NULL,' \19            'email_address TEXT NOT NULL,' \20            'phone_number TEXT NOT NULL)'21    cursor.execute(query)22    query = 'CREATE TABLE IF NOT EXISTS book (' \23            'isbn TEXT PRIMARY KEY NOT NULL,' \24            'title TEXT NOT NULL,' \25            'authors TEXT NOT NULL,' \26            'publisher TEXT NOT NULL,' \27            'publish_year INT NOT NULL)'28    cursor.execute(query)29    query = 'CREATE TABLE IF NOT EXISTS lend (' \30            'nim TEXT NOT NULL,' \31            'isbn TEXT NOT NULL,' \32            'lend_date INT NOT NULL,' \33            'return_date INT NOT NULL,' \34            'FOREIGN KEY (nim) REFERENCES user (nim),' \35            'FOREIGN KEY (isbn) REFERENCES book (isbn))'36    # tanggal pinjam dan kembali disimpan dalam format UNIX epoch (detik sejak 1 Jan 1970, 00:00:00 UTC)37    cursor.execute(query)38    con.commit()39    __close_db(cursor, con)40def get_all_user():41    con = __open_db()42    cursor = con.cursor()43    result_set = cursor.execute('SELECT * FROM user')44    result = []45    for nim, name, password, email_address, phone_number in result_set:46        result.append(User(nim, name, password, email_address, phone_number))47    __close_db(cursor, con)48    return result49def get_user(nim):  # Get user by NIM (Primary Key)50    con = __open_db()51    cursor = con.cursor()52    result_set = cursor.execute('SELECT * FROM user WHERE nim=?', [nim])53    result = []54    for nim, name, password, email_address, phone_number in result_set:55        result.append(User(nim, name, password, email_address, phone_number))56    __close_db(cursor, con)57    return result58def edit_user(nim, email_address, phone_number):59    data = email_address, phone_number, nim60    con = __open_db()61    cursor = con.cursor()62    cursor.execute('UPDATE user SET email_address=?, phone_number=? WHERE nim=?', data)63    con.commit()64    __close_db(cursor, con)65def get_all_book():66    con = __open_db()67    cursor = con.cursor()68    result_set = cursor.execute('SELECT * FROM book')69    result = []70    for isbn, title, authors, publisher, publish_year in result_set:71        result.append(Book(isbn, title, authors, publisher, publish_year))72    __close_db(cursor, con)73    return result74def search_book(keyword):75    con = __open_db()76    cursor = con.cursor()77    result_set = cursor.execute('SELECT * FROM book')78    result = []79    for isbn, title, authors, publisher, publish_year in result_set:80        if title.find(keyword) == -1:  # Jika judul buku tidak mengandung keyword81            continue82        result.append(Book(isbn, title, authors, publisher, publish_year))83    __close_db(cursor, con)84    return result85def get_book(isbn):  # Get book by ISBN (Primary Key)86    con = __open_db()87    cursor = con.cursor()88    result_set = cursor.execute('SELECT * FROM book WHERE isbn=?', [isbn])89    result = []90    for isbn, title, authors, publisher, publish_year in result_set:91        result.append(Book(isbn, title, authors, publisher, publish_year))92    __close_db(cursor, con)93    return result94def get_lend(nim):  # Get data peminjaman buku by user NIM (peminjam)95    con = __open_db()96    cursor = con.cursor()97    result_set = cursor.execute('SELECT * FROM lend WHERE nim=?', [nim])98    result = []99    for nim, isbn, lend_date, return_date in result_set:100        if return_date < int(time.time()):101            # Tidak dimasukkan ke list hasil select jika waktu pemngembalian buku sudah lewat dari hari ini102            continue103        result.append(Lend(get_user(nim)[0], get_book(isbn)[0], lend_date, return_date))104    __close_db(cursor, con)105    return result106def insert_lend(nim, isbn, lend_date, return_date):  # Insert data peminjaman buku107    data = nim, isbn, lend_date, return_date108    con = __open_db()109    cursor = con.cursor()110    cursor.execute('INSERT INTO lend VALUES(?,?,?,?)', data)111    con.commit()112    __close_db(cursor, con)...

mysql.py

Source:mysql.py

...37        Executes a query, gets all the values and then closes up the connection38        """39        self.cursor.execute(query, values)40        retval = self.cursor.fetchall()41        self.__close_db()42        return retval43    def fetchone_query_and_close(self, query, values):44        """45        Executes a query, gets the first value, and closes up the connection46        """47        self.cursor.execute(query, values)48        retval = self.cursor.fetchone()49        self.__close_db()50        return retval51    def execute_query_and_close(self, query, values):52        """53        Executes a query and closes the connection54        """55        retval = self.cursor.execute(query, values)56        self.__close_db()57        return retval58    def __close_db(self):59        self.cursor.close()...

base.py

Source:base.py

1from functools import wraps2from app import db3import logging4logger = logging.getLogger()5def commit_on_success(func):6    """Decorate any function to commit the session on success, rollback in7    the case of error.8    supports nested usage, committing only on the outermost nesting9    """10    @wraps(func)11    def commit_wrapper(*args, **kwargs):12        if hasattr( db.session, '__commit_on_success_active') and db.session.__commit_on_success_active:13            #don't mess with the commit14            return func(*args, **kwargs)15        db.session.__commit_on_success_active = True16        try:17            result = func(*args, **kwargs)18            db.session.commit()19        except Exception, e:20            db.session.rollback()21            logger.exception(u"Error occured in commit_on_success")22            db.session.__commit_on_success_active = False23            raise e24        else:25            db.session.__commit_on_success_active = False26            return result27    return commit_wrapper28# establish a constraint naming convention.29# see http://docs.sqlalchemy.org/en/latest/core/constraints.html#configuring-constraint-naming-conventions30#31db.metadata.naming_convention={32        "pk": "pk_%(table_name)s",33        "fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",34        "uq": "uq_%(table_name)s_%(column_0_name)s",35        "ix": "ix_%(table_name)s_%(column_0_name)s"36    }37def close_db(func):38    #closes the DB.  used for calls outside of flask, AKA celery39    @wraps(func)40    def wrapper(*args, **kwargs):41        if hasattr( db.session, '__close_db') and db.session.__close_db:42            #don't mess with the commit43            return func(*args, **kwargs)44        db.session.__close_db = True45        try:46            result = func(*args, **kwargs)47            db.session.close()48        except Exception, e:49            logger.exception(u"Error occurred closing db connection")50            db.session.__close_db = False51            raise e52        else:53            db.session.__close_db = False54            return result...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.