Test connection to data source by issuing a trivial query.
(name, organization="default")
| 66 | help="The organization the user belongs to " "(leave blank for 'default').", |
| 67 | ) |
| 68 | def test(name, organization="default"): |
| 69 | """Test connection to data source by issuing a trivial query.""" |
| 70 | try: |
| 71 | org = models.Organization.get_by_slug(organization) |
| 72 | data_source = models.DataSource.query.filter( |
| 73 | models.DataSource.name == name, models.DataSource.org == org |
| 74 | ).one() |
| 75 | print("Testing connection to data source: {} (id={})".format(name, data_source.id)) |
| 76 | try: |
| 77 | data_source.query_runner.test_connection() |
| 78 | except Exception as e: |
| 79 | print("Failure: {}".format(e)) |
| 80 | exit(1) |
| 81 | else: |
| 82 | print("Success") |
| 83 | except NoResultFound: |
| 84 | print("Couldn't find data source named: {}".format(name)) |
| 85 | exit(1) |
| 86 | |
| 87 | |
| 88 | @manager.command() |
no test coverage detected