Create a connection to the MySQL DB.
(self, database_env="test", conf_creds=None)
| 5 | """This class wraps MySQL database methods for easy use.""" |
| 6 | |
| 7 | def __init__(self, database_env="test", conf_creds=None): |
| 8 | """Create a connection to the MySQL DB.""" |
| 9 | import fasteners |
| 10 | import time |
| 11 | from seleniumbase import config as sb_config |
| 12 | from seleniumbase.config import settings |
| 13 | from seleniumbase.core import settings_parser |
| 14 | from seleniumbase.fixtures import constants |
| 15 | from seleniumbase.fixtures import shared_utils |
| 16 | |
| 17 | pip_find_lock = fasteners.InterProcessLock( |
| 18 | constants.PipInstall.FINDLOCK |
| 19 | ) |
| 20 | with pip_find_lock: |
| 21 | try: |
| 22 | import cryptography # noqa: F401 |
| 23 | import pymysql |
| 24 | except Exception: |
| 25 | shared_utils.pip_install("PyMySQL[rsa]", version="1.1.1") |
| 26 | import pymysql |
| 27 | db_server = settings.DB_HOST |
| 28 | db_port = settings.DB_PORT |
| 29 | db_user = settings.DB_USERNAME |
| 30 | db_pass = settings.DB_PASSWORD |
| 31 | db_schema = settings.DB_SCHEMA |
| 32 | if getattr(sb_config, "settings_file", None): |
| 33 | override = settings_parser.set_settings(sb_config.settings_file) |
| 34 | if "DB_HOST" in override.keys(): |
| 35 | db_server = override["DB_HOST"] |
| 36 | if "DB_PORT" in override.keys(): |
| 37 | db_port = override["DB_PORT"] |
| 38 | if "DB_USERNAME" in override.keys(): |
| 39 | db_user = override["DB_USERNAME"] |
| 40 | if "DB_PASSWORD" in override.keys(): |
| 41 | db_pass = override["DB_PASSWORD"] |
| 42 | if "DB_SCHEMA" in override.keys(): |
| 43 | db_schema = override["DB_SCHEMA"] |
| 44 | retry_count = 3 |
| 45 | backoff = 1.2 # Time to wait (in seconds) between retries. |
| 46 | count = 0 |
| 47 | while count < retry_count: |
| 48 | try: |
| 49 | self.conn = pymysql.connect( |
| 50 | host=db_server, |
| 51 | port=db_port, |
| 52 | user=db_user, |
| 53 | password=db_pass, |
| 54 | database=db_schema, |
| 55 | ) |
| 56 | self.conn.autocommit(True) |
| 57 | self.cursor = self.conn.cursor() |
| 58 | return |
| 59 | except Exception: |
| 60 | time.sleep(backoff) |
| 61 | count = count + 1 |
| 62 | if retry_count == 3: |
| 63 | print("Unable to connect to Database after 3 retries.") |
| 64 | raise |