text: from passlib.apps import custom_app_context as pwd_context
  import redash.models
  from redash.models import db
  from redash.permissions import ACCESS_TYPE_MODIFY
  from redash.utils import gen_query_hash, utcnow
  from redash.utils.configuration import ConfigurationContainer
  
  
  class ModelFactory(object):
      def __init__(self, model, **kwargs):
          self.model = model
          self.kwargs = kwargs
  
      def _get_kwargs(self, override_kwargs):
          kwargs = self.kwargs.copy()
          kwargs.update(override_kwargs)
  
          for key, arg in kwargs.items():
              if callable(arg):
                  kwargs[key] = arg()
  
          return kwargs
  
      def create(self, **override_kwargs):
          kwargs = self._get_kwargs(override_kwargs)
          obj = self.model(**kwargs)
          db.session.add(obj)
          db.session.commit()
          return obj
  
  
  class Sequence(object):
      def __init__(self, string):
          self.sequence = 0
          self.string = string
  
      def __call__(self):
          self.sequence += 1
  
          return self.string.format(self.sequence)
  
  
  user_factory = ModelFactory(
      redash.models.User,
      name="John Doe",
      email=Sequence("test{}@example.com"),
      password_hash=pwd_context.encrypt("test1234"),
      group_ids=[2],
      org_id=1,
  )
  
  org_factory = ModelFactory(
      redash.models.Organization,
      name=Sequence("Org {}"),
      slug=Sequence("org{}.example.com"),
      settings={},
  )
  
  data_source_factory = ModelFactory(
      redash.models.DataSource,
      name=Sequence("Test {}"),
      type="pg",
      # If we don't use lambda here it will reuse the same options between tests:
      options=lambda: ConfigurationContainer.from_json('{"dbname": "test"}'),
      org_id=1,
  )
  
  dashboard_factory = ModelFactory(
      redash.models.Dashboard,
      name="test",
      user=user_factory.create,
      layout="[]",
      is_draft=False,
      org=1,
  )
  
  api_key_factory = ModelFactory(redash.models.ApiKey, object=dashboard_factory.create)
  
  application_factory = ModelFactory(
      redash.models.Application,
      name="test",
      active=True,
      org=1,
  )
  
  application_dashboard_factory = ModelFactory(
      redash.models.ApplicationDashboard,
  )
  
  query_factory = ModelFactory(
      redash.models.Query,
      name="Query",
      description="",
      query_text="SELECT 1",
      user=user_factory.create,
      is_archived=False,
      is_draft=False,
      schedule=None,
      data_source=data_source_factory.create,
      org_id=1,
  )
  
  query_with_params_factory = ModelFactory(
      redash.models.Query,
      name="New Query with Params",
      description="",
      query_text="SELECT {{param1}}",
      user=user_factory.create,
      is_archived=False,
      is_draft=False,
      schedule={},
      data_source=data_source_factory.create,
      org_id=1,
  )
  
  access_permission_factory = ModelFactory(
      redash.models.AccessPermission,
      object_id=query_factory.create,
      object_type=redash.models.Query.__name__,
      access_type=ACCESS_TYPE_MODIFY,
      grantor=user_factory.create,
      grantee=user_factory.create,
  )
  
  alert_factory = ModelFactory(
      redash.models.Alert,
      name=Sequence("Alert {}"),
      query_rel=query_factory.create,
      user=user_factory.create,
      options={},
  )
  
  query_result_factory = ModelFactory(
      redash.models.QueryResult,
      data='{"columns":{}, "rows":[]}',
      runtime=1,
      retrieved_at=utcnow,
      query_text="SELECT 1",
      query_hash=gen_query_hash("SELECT 1"),
      data_source=data_source_factory.create,
      org_id=1,
  )
  
  visualization_factory = ModelFactory(
      redash.models.Visualization,
      type="CHART",
      query_rel=query_factory.create,
      name="Chart",
      description="",
      options="{}",
  )
  
  widget_factory = ModelFactory(
      redash.models.Widget,
      width=1,
      options="{}",
      dashboard=dashboard_factory.create,
      visualization=visualization_factory.create,
  )
  
  destination_factory = ModelFactory(
      redash.models.NotificationDestination,
      org_id=1,
      user=user_factory.create,
      name=Sequence("Destination {}"),
      type="slack",
      options=lambda: ConfigurationContainer.from_json('{"url": "https://www.slack.com"}'),
  )
  
  alert_subscription_factory = ModelFactory(
      redash.models.AlertSubscription,
      user=user_factory.create,
      destination=destination_factory.create,
      alert=alert_factory.create,
  )
  
  query_snippet_factory = ModelFactory(
      redash.models.QuerySnippet,
      trigger=Sequence("trigger {}"),
      description="description",
      snippet="snippet",
  )
  
  
  class Factory(object):
      def __init__(self):
          self.org, self.admin_group, self.default_group = redash.models.init_db()
          self._data_source = None
          self._user = None
  
      @property
      def user(self):
          if self._user is None:
              self._user = self.create_user()
              # Test setup creates users, they need to be in the db by the time
              # the handler's db transaction starts.
              db.session.commit()
          return self._user
  
      @property
      def data_source(self):
          if self._data_source is None:
              self._data_source = data_source_factory.create(org=self.org)
              db.session.add(
                  redash.models.DataSourceGroup(
                      group=self.default_group, data_source=self._data_source
                  )
              )
  
          return self._data_source
  
      def create_org(self, **kwargs):
          org = org_factory.create(**kwargs)
          self.create_group(
              org=org, type=redash.models.Group.BUILTIN_GROUP, name="default"
          )
          self.create_group(
              org=org,
              type=redash.models.Group.BUILTIN_GROUP,
              name="admin",
              permissions=["admin"],
          )
  
          return org
  
      def create_user(self, **kwargs):
          args = {"org": self.org, "group_ids": [self.default_group.id]}
  
          if "org" in kwargs:
              args["group_ids"] = [kwargs["org"].default_group.id]
  
          args.update(kwargs)
          return user_factory.create(**args)
  
      def create_admin(self, **kwargs):
          args = {
              "org": self.org,
              "group_ids": [self.admin_group.id, self.default_group.id],
          }
  
          if "org" in kwargs:
              args["group_ids"] = [
                  kwargs["org"].default_group.id,
                  kwargs["org"].admin_group.id,
              ]
  
          args.update(kwargs)
          return user_factory.create(**args)
  
      def create_group(self, **kwargs):
          args = {"name": "Group", "org": self.org}
  
          args.update(kwargs)
  
          g = redash.models.Group(**args)
          return g
  
      def create_alert(self, **kwargs):
          args = {"user": self.user, "query_rel": self.create_query()}
  
          args.update(**kwargs)
          return alert_factory.create(**args)
  
      def create_alert_subscription(self, **kwargs):
          args = {"user": self.user, "alert": self.create_alert()}
  
          args.update(**kwargs)
          return alert_subscription_factory.create(**args)
  
      def create_data_source(self, **kwargs):
          group = None
          if "group" in kwargs:
              group = kwargs.pop("group")
          args = {"org": self.org}
          args.update(kwargs)
  
          if group and "org" not in kwargs:
              args["org"] = group.org
  
          view_only = args.pop("view_only", False)
          data_source = data_source_factory.create(**args)
  
          if group:
              db.session.add(
                  redash.models.DataSourceGroup(
                      group=group, data_source=data_source, view_only=view_only
                  )
              )
  
          return data_source
  
      def create_dashboard(self, **kwargs):
          args = {"user": self.user, "org": self.org}
          args.update(kwargs)
          return dashboard_factory.create(**args)
  
      def create_access_token(self, **kwargs):
          token = redash.models.AccessToken()
          access_token = token.new(300)
          return access_token
  
      def create_query(self, **kwargs):
          args = {"user": self.user, "data_source": self.data_source, "org": self.org}
          args.update(kwargs)
          return query_factory.create(**args)
  
      def create_query_with_params(self, **kwargs):
          args = {"user": self.user, "data_source": self.data_source, "org": self.org}
          args.update(kwargs)
          return query_with_params_factory.create(**args)
  
      def create_access_permission(self, **kwargs):
          args = {"grantor": self.user}
          args.update(kwargs)
          return access_permission_factory.create(**args)
  
      def create_query_result(self, **kwargs):
          args = {"data_source": self.data_source}
  
          args.update(kwargs)
  
          if "data_source" in args and "org" not in args:
              args["org"] = args["data_source"].org
  
          return query_result_factory.create(**args)
  
      def create_visualization(self, **kwargs):
          args = {"query_rel": self.create_query()}
          args.update(kwargs)
          return visualization_factory.create(**args)
  
      def create_visualization_with_params(self, **kwargs):
          args = {"query_rel": self.create_query_with_params()}
          args.update(kwargs)
          return visualization_factory.create(**args)
  
      def create_widget(self, **kwargs):
          args = {
              "dashboard": self.create_dashboard(),
              "visualization": self.create_visualization(),
          }
          args.update(kwargs)
          return widget_factory.create(**args)
  
      def create_api_key(self, **kwargs):
          args = {"org": self.org}
          args.update(kwargs)
          return api_key_factory.create(**args)
  
      def create_application(self, **kwargs):
          args = {"org": self.org}
          args.update(kwargs)
          return application_factory.create(**args)
  
      def create_application_dashboard(self, **kwargs):
          args = {"created_by_id": self.user.id}
          args.update(kwargs)
          return application_dashboard_factory.create(**args)
  
      def create_destination(self, **kwargs):
          args = {"org": self.org}
          args.update(kwargs)
          return destination_factory.create(**args)
  
      def create_query_snippet(self, **kwargs):
          args = {"user": self.user, "org": self.org}
          args.update(kwargs)
          return query_snippet_factory.create(**args)
  