| 27 | |
| 28 | |
| 29 | class Query(models.Model): |
| 30 | title = models.CharField(max_length=255) |
| 31 | sql = models.TextField(blank=False, null=False) |
| 32 | description = models.TextField(blank=True) |
| 33 | created_by_user = models.ForeignKey( |
| 34 | settings.AUTH_USER_MODEL, |
| 35 | null=True, |
| 36 | blank=True, |
| 37 | on_delete=models.CASCADE |
| 38 | ) |
| 39 | created_at = models.DateTimeField(auto_now_add=True) |
| 40 | last_run_date = models.DateTimeField(auto_now=True) |
| 41 | snapshot = models.BooleanField( |
| 42 | default=False, |
| 43 | help_text=_("Include in snapshot task (if enabled)") |
| 44 | ) |
| 45 | # NOTE this field is deprecated in favor of database_connection and no longer in use. |
| 46 | # It is present in the 6.0 release to preserve backwards compatibility in case there is need for a rollback. |
| 47 | # It will be removed in a future release (e.g. v6.1) |
| 48 | connection = models.CharField( |
| 49 | blank=True, |
| 50 | max_length=128, |
| 51 | default="", |
| 52 | help_text=_( |
| 53 | "Name of DB connection (as specified in settings) to use for " |
| 54 | "this query." |
| 55 | "Will use EXPLORER_DEFAULT_CONNECTION if left blank" |
| 56 | ) |
| 57 | ) |
| 58 | database_connection = models.ForeignKey(to=DatabaseConnection, on_delete=models.SET_NULL, null=True) |
| 59 | few_shot = models.BooleanField(default=False, help_text=_( |
| 60 | "Will be included as a good example of SQL in assistant queries that use relevant tables")) |
| 61 | |
| 62 | def __init__(self, *args, **kwargs): |
| 63 | self.params = kwargs.get("params") |
| 64 | kwargs.pop("params", None) |
| 65 | super().__init__(*args, **kwargs) |
| 66 | |
| 67 | class Meta: |
| 68 | ordering = ["title"] |
| 69 | verbose_name = _("Query") |
| 70 | verbose_name_plural = _("Queries") |
| 71 | |
| 72 | def __str__(self): |
| 73 | return str(self.title) |
| 74 | |
| 75 | def get_run_count(self): |
| 76 | return self.querylog_set.count() |
| 77 | |
| 78 | def last_run_log(self): |
| 79 | ql = self.querylog_set.first() |
| 80 | return ql or QueryLog(success=True, run_at=self.created_at) |
| 81 | |
| 82 | def avg_duration_display(self): |
| 83 | d = self.avg_duration() |
| 84 | if d: |
| 85 | return f"{self.avg_duration():10.3f}" |
| 86 | return "" |
no outgoing calls