Source code for catcher_modules.database.mssql
from catcher.steps.external_step import ExternalStep
from catcher.steps.step import update_variables
from catcher_modules.utils import db_utils
from catcher_modules.database import SqlAlchemyDb
[docs]class MSSql(ExternalStep, SqlAlchemyDb):
"""
Allows you to run queries on Microsoft SQL `Server <https://www.microsoft.com/en-us/sql-server/>`_.
:Input:
:conf: mssql 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
- driver: odbc driver name you've installed. *Optional* If not specified, the default driver, which comes with
catcher-modules Dockerfile will be used.
: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
::
mssql:
request:
conf:
dbname: test
user: user
password: password
host: localhost
port: 1433
driver: ODBC Driver 17 for SQL Server
sql: 'select count(*) as count from test'
register: {documents: '{{ OUTPUT }}'}
**Note** that we alias count. For some reason sqlalchemy for mssql will return `count(*)` as a column name
instead of `count`.
Insert into test, using string configuration
::
mssql:
request:
conf: 'user:password@localhost:5432/test'
sql: 'insert into test(id, num) values(3, 3);'
Insert into test, using string configuration with pymssql (pymssql should be installed)
::
mssql:
request:
conf: 'mssql+pymssql://user:password@localhost:5432/test'
sql: 'insert into test(id, num) values(3, 3);'
"""
@property
def dialect(self) -> str:
return "mssql+pyodbc"
def get_engine(self, conf):
if isinstance(conf, dict):
driver = 'ODBC Driver 17 for SQL Server' if 'driver' not in conf else conf['driver']
else:
driver = None
return db_utils.get_engine(conf, self.dialect, driver)
@update_variables
def action(self, includes: dict, variables: dict) -> any:
body = self.simple_input(variables)
return variables, self.execute(body, variables)