Source code for catcher_modules.database.mysql

from catcher.steps.external_step import ExternalStep
from catcher.steps.step import update_variables

from catcher_modules.database import SqlAlchemyDb


[docs]class MySql(ExternalStep, SqlAlchemyDb): """ Allows you to run queries on `MySQL <https://www.mysql.com/>`_ (and all mysql compatible databases like `MariaDB <https://mariadb.org/>`_). :Input: :conf: mysql configuration. Can be a single line string or object. Dialect is not mandatory. **Required**. - dbname: name of the database to connect to - user: database user - host: database host - password: user's password - port: database port :query: query to run. **Deprecated since 5.2** :sql: query or sql file from resources to run. **Required** :Examples: Select all from test, use object configuration :: mysql: request: conf: dbname: test user: user password: password host: localhost port: 3306 sql: 'select count(*) as count from test' register: {documents: '{{ OUTPUT }}'} **Note** that we alias count. For some reason sqlalchemy for mysql will return `count(*)` as a column name instead of `count`. Insert into test, using string configuration :: mysql: request: conf: 'user:password@localhost:3306/test' sql: 'insert into test(id, num) values(3, 3);' Insert into test, using string configuration with dialect :: mysql: request: conf: 'mysql+pymysql://user:password@localhost:3306/test' sql: 'insert into test(id, num) values(3, 3);' """ @property def dialect(self) -> str: return "mysql+pymysql" @update_variables def action(self, includes: dict, variables: dict) -> any: body = self.simple_input(variables) return variables, self.execute(body, variables)