# pylint: disable=W0613,C0115,C0302
"""All models for the OpenAI Function Calling API app."""
import logging
import warnings
from functools import cached_property
from typing import Any, List, Optional, Type
from urllib.parse import ParseResult, urljoin, urlparse
from django.core.exceptions import ValidationError
from django.db import models
from django.http import HttpRequest
from django.urls import reverse
from rest_framework import serializers
# our stuff
from smarter.apps.account.models import (
Account,
MetaDataWithOwnershipModel,
User,
UserProfile,
)
from smarter.apps.account.serializers import UserProfileSerializer
from smarter.apps.account.utils import (
account_number_from_url,
get_cached_admin_user_for_account,
smarter_cached_objects,
)
from smarter.apps.plugin.manifest.controller import PluginController
from smarter.apps.plugin.manifest.models.common.plugin.model import SAMPluginCommon
from smarter.apps.plugin.models import PluginMeta
from smarter.apps.plugin.plugin.base import PluginBase
from smarter.apps.provider.models import Provider
from smarter.common.conf import smarter_settings
from smarter.common.const import SmarterEnvironments
from smarter.common.exceptions import SmarterValueError
from smarter.common.helpers.console_helpers import (
formatted_text,
formatted_text_green,
formatted_text_red,
)
from smarter.common.helpers.llm import get_date_time_string
from smarter.common.utils import rfc1034_compliant_str, smarter_build_absolute_uri
from smarter.lib import json
from smarter.lib.cache import cache_results
from smarter.lib.cache import lazy_cache as cache
from smarter.lib.django import waffle
from smarter.lib.django.models import TimestampedModel
from smarter.lib.django.request import SmarterRequestMixin
from smarter.lib.django.validators import SmarterValidator
from smarter.lib.django.waffle import SmarterWaffleSwitches
from smarter.lib.drf.models import SmarterAuthToken
from smarter.lib.logging import WaffleSwitchedLoggerWrapper
from smarter.lib.manifest.loader import SAMLoader
from .signals import (
chatbot_deploy,
chatbot_deploy_failed,
chatbot_deploy_status_changed,
chatbot_dns_failed,
chatbot_dns_verification_initiated,
chatbot_dns_verification_status_changed,
chatbot_dns_verified,
chatbot_undeploy,
)
CACHE_PREFIX = "ChatBotHelper_"
def should_log(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.CHATBOT_LOGGING)
def should_log_chatbot_helper(level):
"""Check if logging should be done based on the waffle switch."""
return waffle.switch_is_active(SmarterWaffleSwitches.CHATBOT_HELPER_LOGGING)
base_logger = logging.getLogger(__name__)
logger = WaffleSwitchedLoggerWrapper(base_logger, should_log)
chatbot_helper_logger = WaffleSwitchedLoggerWrapper(base_logger, should_log_chatbot_helper)
# -----------------------------------------------------------------------------
# ChatBot Models. These implement a ChatBot API for a customer account.
# -----------------------------------------------------------------------------
[docs]
class ChatBotCustomDomain(MetaDataWithOwnershipModel):
"""
Represents a DNS host record for a customer account's ChatBot, linked to an AWS Hosted Zone.
This model is used to manage custom domains for chatbots within the Smarter platform. Each instance
of this model corresponds to a DNS host (subdomain) that is associated with a specific customer
account and is managed through AWS Route 53 Hosted Zones.
The primary purpose of this model is to enable customers to use their own branded domains for
chatbot endpoints, rather than relying solely on default platform-provided domains. This allows
for improved branding, trust, and integration with customer infrastructure.
**Key Features**
- Associates a custom domain with a customer :class:`Account`.
- Stores the AWS Hosted Zone ID for DNS management and automation.
- Tracks the verification status of the domain, indicating whether DNS records have been correctly
configured and validated.
- Supports caching of verified domains for efficient lookup and validation across the platform.
**Usage Scenarios**
- When a customer wishes to deploy a chatbot at a custom subdomain (e.g., ``chatbot.example.com``),
an instance of this model is created to represent and manage that domain.
- The platform uses the AWS Hosted Zone ID to automate DNS record creation and validation as part
of the chatbot deployment workflow.
- The ``is_verified`` field is updated as part of the DNS verification process, ensuring that only
properly configured domains are used for chatbot endpoints.
**Integration**
- This model is referenced by other chatbot-related models, such as :class:`ChatBot` and
:class:`ChatBotCustomDomainDNS`, to provide a complete mapping between chatbots, their domains,
and DNS records.
- The platform uses this model to enforce domain uniqueness and to prevent conflicts between
customer accounts.
**Notes**
- The domain name must be a valid DNS hostname and is validated upon saving.
- Caching is used to optimize the retrieval of verified domains, reducing database load and
improving performance for domain-related checks.
- This model is intended for internal use within the Smarter platform and is not exposed directly
to end users.
**Example**
.. code-block:: python
# Create a new custom domain for a chatbot
custom_domain = ChatBotCustomDomain.objects.create(
account=my_account,
aws_hosted_zone_id="Z1234567890ABCDEF",
domain_name="chatbot.example.com",
is_verified=False,
)
# Retrieve all verified custom domains
verified_domains = ChatBotCustomDomain.get_verified_domains()
"""
class Meta:
verbose_name_plural = "ChatBot Custom Domains"
#: The AWS Hosted Zone ID associated with this custom domain. This ID is used for DNS management via AWS Route 53.
#: Example: "Z1234567890ABCDEF"
aws_hosted_zone_id = models.CharField(max_length=255)
#: The custom domain name for the ChatBot. This should be a valid DNS hostname.
#: Example: "chatbot.example.com"
domain_name = models.CharField(max_length=255)
#: Indicates whether the custom domain has been verified. A verified domain has the correct DNS records configured.
#: This is managed by the asynchronous ChatBot deployment process.
#: Example: True
is_verified = models.BooleanField(default=False, blank=True, null=True)
[docs]
@classmethod
def get_verified_domains(cls):
"""
Get all verified custom domains from cache or database.
:returns: List of verified domain names.
:rtype: List[str]
"""
# Try to get the list from cache
cache_key = "ChatBotCustomDomain_chatbot_verified_custom_domains"
verified_domains = cache.get(cache_key)
# If the list is not in cache, fetch it from the database
if not verified_domains:
verified_domains = list(cls.objects.filter(is_verified=True).values_list("domain_name", flat=True))
cache.set(key=cache_key, value=verified_domains, timeout=smarter_settings.cache_expiration)
if waffle.switch_is_active(SmarterWaffleSwitches.CACHE_LOGGING):
logger.debug("get_verified_domains() caching %s", cache_key)
return verified_domains
[docs]
def save(self, *args, **kwargs):
"""
Save the ChatBotCustomDomain instance, validating the domain name.
:raises ValidationError: If the domain name is not valid.
:returns: None
"""
if self.domain_name:
SmarterValidator.validate_domain(self.domain_name)
super().save(*args, **kwargs)
def __str__(self):
return str(self.domain_name) if self.domain_name else "undefined"
[docs]
class ChatBotCustomDomainDNS(TimestampedModel):
"""
Represents a DNS record associated with a custom domain for a ChatBot within the Smarter platform.
This model is responsible for storing and managing individual DNS records that are linked to a
specific :class:`ChatBotCustomDomain`. Each instance of this model corresponds to a single DNS
record, such as an A, CNAME, or TXT record, which is required for the proper operation and
verification of a chatbot's custom domain.
The primary use case for this model is to facilitate the automation and tracking of DNS
configurations necessary for deploying chatbots on customer-branded domains. By maintaining a
record of all DNS entries related to a chatbot's custom domain, the platform can automate DNS
verification, support trouble shooting, and ensure that all required DNS records are present and
correctly configured.
**Key Features**
- Associates each DNS record with a specific :class:`ChatBotCustomDomain`.
- Stores the record name, type (such as A, CNAME, TXT), value, and TTL (time-to-live).
- Supports management of multiple DNS records per custom domain, enabling complex DNS setups.
- Facilitates DNS verification workflows and integration with external DNS providers (e.g., AWS Route 53).
**Usage Scenarios**
- When deploying a chatbot to a custom domain, instances of this model are created to represent
the required DNS records (e.g., for domain verification, routing, or certificate issuance).
- The platform can query this model to retrieve all DNS records associated with a given custom domain,
enabling automated checks and updates.
- Used internally by deployment and verification processes to track the status and configuration
of DNS records for each chatbot custom domain.
**Integration**
- Closely linked to :class:`ChatBotCustomDomain`, providing a one-to-many relationship between
a custom domain and its DNS records.
- Referenced by deployment, verification, and trouble shooting workflows within the Smarter platform.
**Notes**
- This model is intended for internal use and is not exposed directly to end users.
- The record fields are validated to ensure compliance with DNS standards.
- TTL defaults to 600 seconds but can be customized as needed for specific DNS requirements.
**Example**
.. code-block:: python
# Create a new DNS record for a chatbot custom domain
dns_record = ChatBotCustomDomainDNS.objects.create(
custom_domain=my_custom_domain,
record_name="_acme-challenge.chatbot.example.com",
record_type="TXT",
record_value="abc123xyz",
record_ttl=600,
)
# Retrieve all DNS records for a custom domain
records = ChatBotCustomDomainDNS.objects.filter(custom_domain=my_custom_domain)
"""
class Meta:
verbose_name_plural = "ChatBot Custom Domain DNS"
#: The ChatBotCustomDomain that this DNS record is associated with.
#: Example: ChatBotCustomDomain(id=1, domain="example.com")
custom_domain = models.ForeignKey(ChatBotCustomDomain, on_delete=models.CASCADE)
#: The name of the DNS record (e.g., "_acme-challenge.chatbot.example.com").
#: Example: "_acme-challenge.chatbot.example.com"
record_name = models.CharField(max_length=255)
#: The type of DNS record (e.g., "A", "CNAME", "TXT").
#: Example: "TXT"
record_type = models.CharField(max_length=255)
#: The value of the DNS record (e.g., "abc123xyz").
#: Example: "abc123xyz"
record_value = models.CharField(max_length=255)
#: The time-to-live (TTL) for the DNS record, in seconds.
#: Example: 600
record_ttl = models.IntegerField(default=600, blank=True, null=True)
def validate_provider(value):
"""
Validate that the provider is in the list of valid chat providers.
:param value: The provider value to validate.
:raises ValidationError: If the provider is not valid.
:returns: None
"""
# pylint: disable=C0415
from smarter.apps.prompt.providers.providers import chat_providers
if not value in chat_providers.all:
raise ValidationError(
"%(value)s is not a valid provider. Valid providers are: %(providers)s",
params={"value": value, "providers": str(chat_providers.all)},
)
[docs]
class ChatBot(MetaDataWithOwnershipModel):
"""
Implements the ChatBot API model for a customer account.
This Django model represents a chatbot instance associated with a specific customer account.
It provides configuration, deployment status, domain management, and API endpoint properties
for each chatbot. The model supports multiple modes of operation (sandbox, custom, default),
DNS verification, TLS certificate management, and integration with external providers.
**Key Features**
- Associates each chatbot with a customer :class:`Account`.
- Supports custom domains and DNS verification via :class:`ChatBotCustomDomain`.
- Tracks deployment status, TLS certificate issuance, and DNS verification.
- Configures default provider, model, system role, temperature, and max tokens.
- Provides properties for generating RFC 1034-compliant names, hosts, and URLs.
- Supports sandbox, default, and custom domain modes.
- Integrates with Django signals for deployment and verification events.
- Serializes chatbot configuration for API and frontend consumption.
**Usage Example**
.. code-block:: python
chatbot = ChatBot.objects.get(account=my_account, name="example")
if chatbot.ready():
print(chatbot.url_chatbot)
**Signals**
- Emits signals on deployment, DNS verification, and certificate status changes.
**See Also**
- :class:`ChatBotCustomDomain`
- :class:`ChatBotCustomDomainDNS`
- :class:`ChatBotPlugin`
- :class:`ChatBotAPIKey`
- :class:`ChatBotFunctions`
- :class:`ChatBotRequests`
:raises SmarterValueError: If invalid URLs or domains are provided.
:raises ValidationError: If provider is not valid.
"""
class Meta:
verbose_name_plural = "ChatBots"
unique_together = ("user_profile", "name")
[docs]
class Modes:
"""
ChatBot API Modes. Defines the operational mode of the ChatBot instance.
Also affects the url scheme and hostname used to access the ChatBot API.
"""
SANDBOX = "sandbox"
CUSTOM = "custom"
DEFAULT = "default"
UNKNOWN = "unknown"
[docs]
class Schemes:
"""ChatBot API Schemes"""
HTTP = "http"
HTTPS = "https"
[docs]
class DnsVerificationStatusChoices(models.TextChoices):
"""
DNS Verification Status Choices for ChatBot Custom Domains.
This is managed by the asynchronous ChatBot deployment process.
"""
VERIFYING = "Verifying", "Verifying"
NOT_VERIFIED = "Not Verified", "Not Verified"
VERIFIED = "Verified", "Verified"
FAILED = "Failed", "Failed"
[docs]
class TlsCertificateIssuanceStatusChoices(models.TextChoices):
"""
TLS Certificate Issuance Status Choices for ChatBot Custom Domains.
This is managed by the asynchronous ChatBot deployment process.
"""
NO_CERTIFICATE = "No Certificate", "No Certificate"
REQUESTED = "Requested", "Requested"
ISSUED = "Issued", "Issued"
FAILED = "Failed", "Failed"
#: The subdomain DNS record associated with this ChatBot.
#: Example: ChatBotCustomDomainDNS(id=1, domain="my-chatbot.example.com")
subdomain = models.ForeignKey(ChatBotCustomDomainDNS, on_delete=models.CASCADE, blank=True, null=True)
#: The custom domain associated with this ChatBot.
#: Example: ChatBotCustomDomain(id=1, domain="example.com")
custom_domain = models.ForeignKey(ChatBotCustomDomain, on_delete=models.CASCADE, blank=True, null=True)
#: Indicates whether the ChatBot is deployed and accessible via its custom or default domain.
#: Modifying this value triggers asynchronous deployment or undeployment processes.
#: Example: True
deployed = models.BooleanField(default=False, blank=True, null=True)
#: The Smarter Provider for the ChatBot's language model.
#: Example: "openai"
provider = models.CharField(
default=smarter_settings.llm_default_provider,
max_length=255,
blank=True,
null=True,
validators=[validate_provider],
)
#: The default language model used by the ChatBot.
#: Example: "gpt-4o-mini"
default_model = models.CharField(max_length=255, blank=True, null=True)
#: The default system role prompt for the ChatBot.
#: Example: "You are a helpful assistant."
default_system_role = models.TextField(default=smarter_settings.llm_default_system_role, blank=True, null=True)
#: The default temperature setting for the ChatBot's language model.
#: Example: 0.7
default_temperature = models.FloatField(default=smarter_settings.llm_default_temperature, blank=True, null=True)
#: The default maximum tokens for the ChatBot's language model responses.
#: Example: 1024
default_max_tokens = models.IntegerField(default=smarter_settings.llm_default_max_tokens, blank=True, null=True)
#: The ChatBot UI configuration fields. Appears in the title bar of the Smarter React ChatBot component.
#: Example: "Stackademy Support Bot"
app_name = models.CharField(default="chatbot", max_length=255, blank=True, null=True)
#: The ChatBot UI configuration fields. Appears in the text input area placeholder.
#: Example: "Stan"
app_assistant = models.CharField(default="Smarter", max_length=255, blank=True, null=True)
#: The ChatBot UI configuration fields. Appears in the welcome message area of the Smarter React ChatBot component.
#: Example: "Welcome to Stackademy!"
app_welcome_message = models.CharField(default="Welcome to the chatbot!", max_length=255, blank=True, null=True)
#: The ChatBot UI configuration fields. Example prompts shown to the user in the Smarter React ChatBot component.
#: Example: ["What AI courses do you offer?", "Is your program free?"]
app_example_prompts = models.JSONField(
default=list,
blank=True,
null=True,
encoder=json.SmarterJSONEncoder,
)
#: The ChatBot UI configuration fields. Placeholder text in the chat input area.
#: Example: "Ask me anything about Stackademy..."
app_placeholder = models.CharField(default="Type something here...", max_length=255, blank=True, null=True)
#: The ChatBot UI configuration fields. URL to the app information button in the top-right
#: of the Smarter React ChatBot component.
#: Example: "https://smarter.sh"
app_info_url = models.URLField(default="https://smarter.sh", blank=True, null=True)
#: The ChatBot UI configuration fields. URL to the app background image in the Smarter React ChatBot component.
#: Example: "https://cdn.smarter.sh/chat-ui/background.png"
app_background_image_url = models.URLField(blank=True, null=True)
#: The ChatBot UI configuration fields. URL to the app logo image in the Smarter React ChatBot component.
#: Example: "https://cdn.smarter.sh/chat-ui/logo.png"
app_logo_url = models.URLField(blank=True, null=True)
#: The ChatBot UI configuration fields. Enables or disables file attachment feature in the Smarter React ChatBot component.
#: Example: True
app_file_attachment = models.BooleanField(default=False, blank=True, null=True)
# : The DNS verification status of the ChatBot's custom domain. This is part of the deployment process and is managed by
# : the asynchronous ChatBot deployment workflow.
# : Example: "Verified"
dns_verification_status = models.CharField(
max_length=255,
default=DnsVerificationStatusChoices.NOT_VERIFIED,
blank=True,
null=True,
choices=DnsVerificationStatusChoices.choices,
)
# : The TLS certificate issuance status of the ChatBot's custom domain. This is part of the deployment process and is managed by
# : the asynchronous ChatBot deployment workflow.
# : Example: "Issued"
tls_certificate_issuance_status = models.CharField(
max_length=255,
default=TlsCertificateIssuanceStatusChoices.NO_CERTIFICATE,
blank=True,
null=True,
choices=TlsCertificateIssuanceStatusChoices.choices,
)
def __str__(self):
return self.url if self.url else "undefined"
@property
def rfc1034_compliant_name(self) -> str:
"""
Returns a RFC 1034 compliant name for the ChatBot. This name is used
in the hostname of the ChatBot's default and custom URLs. The name
is constructed by combining the ChatBot's name and the username of
the associated user profile, separated by a dot. The resulting name
adheres to the following rules:
- lower case
- alphanumeric characters and hyphens only [a-z0-9-]
- starts and ends with an alphanumeric character
- max length of 63 characters
- no consecutive hyphens
- no leading or trailing hyphens
- no underscores or special characters
- no spaces
- no dots except for separating the ChatBot name and username
- no more than one dot
Examples:
- For a ChatBot with name "example" and associated user profile "adminuser",
that IS the account admin, the resulting RFC 1034 compliant name
would be "example"
- For a ChatBot with name "example" and associated user profile "user123",
that is NOT the account admin, the resulting RFC 1034 compliant name
would be "example.user123"
:returns: RFC 1034 compliant name
:rtype: str
"""
user_profile: UserProfile = self.user_profile
admin_user = get_cached_admin_user_for_account(account=user_profile.cached_account) # type: ignore[arg-type]
if user_profile.cached_user == admin_user:
raw_str = self.name
else:
# note: rfc1034_compliant_str() filters out the "."
raw_str = f"{self.name}-{user_profile.user.username}"
return rfc1034_compliant_str(raw_str)
@property
def default_system_role_enhanced(self):
"""
prepends a date/time string to the default_system_role
Example: "2024-06-01 12:00:00 System: You are a helpful assistant."
:returns: enhanced system role string
:rtype: str
"""
return f"{get_date_time_string()}{self.default_system_role}"
@property
def base_api_domain(self):
"""
The base API domain for the ChatBot. This is the domain that is used in the default hostname for the ChatBot.
Examples:
example 1. given:
- environment is "alpha"
- environment API domain "alpha.api.example.com"
the resulting base API domain would be: 'alpha.api.example.com'
example 2. given:
- environment is "local"
- environment API domain "api.localhost:9357"
the resulting base API domain would be: 'api.local.example.com'
"""
if smarter_settings.environment in SmarterEnvironments.aws_environments:
return smarter_settings.environment_api_domain
return smarter_settings.proxy_api_domain
@property
def base_default_host(self):
"""
The base default hostname for the ChatBot. This is the part of the hostname
that comes after the RFC 1034 compliant name. It includes the account number
and the environment API domain.
Examples:
example 1. given:
- a ChatBot associated with an account number "1234-5678-9012"
- environment is "alpha"
- environment API domain "alpha.api.example.com"
the resulting base default host would be: '.1234-5678-9012.alpha.api.example.com'
example 2. given:
- a ChatBot associated with an account number "1234-5678-9012"
- environment is "local"
- environment API domain "api.localhost:9357"
the resulting base default host would be: '.1234-5678-9012.api.local.example.com'
"""
user_profile: UserProfile = self.user_profile
return f"{user_profile.account.account_number}.{self.base_api_domain}"
@property
def default_host(self):
"""
The default hostname for the ChatBot.
Examples:
example 1. given:
- self.name: 'example'
- self.account.account_number: '1234-5678-9012'
- smarter_settings.environment = "alpha"
- smarter_settings.environment_api_domain: 'alpha.api.example.com'
The domain would be: 'example.1234-5678-9012.alpha.api.example.com'
example 2. given:
- self.name: 'example'
- self.account.account_number: '1234-5678-9012'
- smarter_settings.environment = "local"
- smarter_settings.environment_api_domain: 'api.localhost:9357'
The domain would be: 'example.1234-5678-9012.api.local.example.com'
:returns: default hostname
:rtype: str
"""
domain = f"{self.rfc1034_compliant_name}.{self.base_default_host}"
SmarterValidator.validate_domain(domain)
return domain
@property
def default_url(self):
"""
The default URL for the ChatBot.
example 'https://example.1234-5678-9012.alpha.api.example.com'
:returns: default URL
:rtype: str
"""
return SmarterValidator.urlify(self.default_host, environment=smarter_settings.environment) # type: ignore[return-value]
@property
def custom_host(self):
"""
The custom hostname for the ChatBot.
Examples:
- self.name: 'example'
- self.custom_domain.domain_name: 'example.com'
example 'example.example.com'
:returns: custom hostname
:rtype: str
"""
if self.custom_domain and self.custom_domain.is_verified:
domain = f"{self.rfc1034_compliant_name}.{self.custom_domain.domain_name}"
SmarterValidator.validate_domain(domain)
return domain
return None
@property
def custom_url(self):
"""
The custom URL for the ChatBot.
example 'https://example.example.com'
:returns: custom URL
:rtype: str
"""
if self.custom_host:
return SmarterValidator.urlify(self.custom_host, environment=smarter_settings.environment) # type: ignore[return-value]
return None
@property
def sandbox_host(self):
"""
The sandbox hostname for the ChatBot. This is the hostname that is
used when the ChatBot is in sandbox mode. For example, when the
ChatBot is being used in the Smarter Workbench.
example 'alpha.platform.smarter.sh'
:returns: sandbox hostname
:rtype: str
"""
return smarter_settings.environment_platform_domain
@property
def sandbox_url(self):
"""
The sandbox URL for the ChatBot. This is the URL that is used when
the ChatBot is in sandbox mode. For example, when the ChatBot is
being used in the Smarter Workbench. maps to "<int:chatbot_id>/"
example: 'https://alpha.platform.smarter.sh/workbench/chatbots/<str:hashed_id>/'
:returns: sandbox URL
:rtype: str
"""
# pylint: disable=C0415
from smarter.apps.prompt.urls import PromptReverseViews
path = reverse(f"{PromptReverseViews.namespace}:{PromptReverseViews.prompt_landing_by_hashed_id}", kwargs={"hashed_id": self.hashed_id}) # type: ignore[arg-type]
url = urljoin(smarter_settings.environment_url, path)
url = SmarterValidator.urlify(url, environment=smarter_settings.environment) # type: ignore[return-value]
return url
@property
def hostname(self):
"""
The hostname for the ChatBot depending on its deployment status.
Returns either the custom hostname (if deployed), the default hostname, or the sandbox hostname.
:returns: hostname
:rtype: str
"""
if self.deployed:
return self.custom_host or self.default_host
return self.sandbox_host
@property
def url(self):
"""
The URL for the ChatBot depending on its deployment status.
example: 'https://my-chatbot.1234-5678-9012.alpha.api.example.com' (custom, deployed)
:returns: URL
:rtype: str
"""
if self.deployed:
return self.custom_url or self.default_url
return self.sandbox_url
@property
def url_chatbot(self):
"""
The Smarter Api url returned by ChatConfigView.config() as the
key, "url_chatbot". This url is consumed by React.js app for http
requests on new prompts.
maps to "<int:chatbot_id>/chat/"
example: "http://localhost:9357/api/v1/chatbots/5174/chat/"
:returns: URL for chatbot API
:rtype: str
"""
# pylint: disable=C0415
from smarter.apps.chatbot.api.v1.urls import ChatBotApiV1ReverseViews
path = reverse(
f"{ChatBotApiV1ReverseViews.namespace}:{ChatBotApiV1ReverseViews.default_chatbot_api_view_by_hashed_id}",
kwargs={"hashed_id": self.hashed_id},
)
url = urljoin(smarter_settings.environment_url, path)
url = SmarterValidator.urlify(url, environment=smarter_settings.environment) # type: ignore[return-value]
if not isinstance(url, str):
raise SmarterValueError("ChatBot.url_chatbot is not a valid string")
return url
@property
def url_chat_config(self):
"""
The Smarter Api url for the Chat config json dict.
The React.js app requests this url during react app startup
to retrieve the UI configuration for the chatbot.
maps to "<int:chatbot_id>/config/"
example: "http://localhost:9357/api/v1/chatbots/5174/config/"
:returns: URL for chatbot config API
:rtype: str
"""
# pylint: disable=C0415
from smarter.apps.chatbot.api.v1.urls import ChatBotApiV1ReverseViews
path = reverse(
f"{ChatBotApiV1ReverseViews.namespace}:{ChatBotApiV1ReverseViews.chat_config_view_by_hashed_id}",
kwargs={"hashed_id": self.hashed_id},
)
url = urljoin(smarter_settings.environment_url, path)
url = SmarterValidator.urlify(url, environment=smarter_settings.environment) # type: ignore[return-value]
return url
@property
def url_chatapp(self) -> str:
"""
(Deprecated) The Smarter Api url for the ChatApp endpoint. This url is used by the React.js app
to load the ChatApp web page.
maps to "chat/"
"""
warnings.warn(
"ChatBot.url_chatapp is deprecated and will be removed in a future release.",
DeprecationWarning,
stacklevel=2,
)
# pylint: disable=C0415
from smarter.apps.prompt.urls import PromptReverseViews
path = reverse(f"{PromptReverseViews.namespace}:{PromptReverseViews.prompt_chat_by_hashed_id}", kwargs={"hashed_id": self.hashed_id}) # type: ignore[arg-type]
url = urljoin(smarter_settings.environment_url, path)
url = SmarterValidator.urlify(url, environment=smarter_settings.environment) # type: ignore[return-value]
return url
@property
def ready(self):
"""
The readiness status of the ChatBot.
A ChatBot is ready if it is its in sandbox mode, or, if it is:
- deployed
- has a verified DNS A record
- has a valid, issued tls certificate.
:returns: readiness status
:rtype: bool
"""
if isinstance(self.url, str) and self.mode(self.url) == self.Modes.SANDBOX:
return True
if self.dns_verification_status != self.DnsVerificationStatusChoices.VERIFIED:
logger.warning(
"ChatBot %s is not ready. DNS verification status is %s",
self.rfc1034_compliant_name,
self.dns_verification_status,
)
return False
if self.tls_certificate_issuance_status != self.TlsCertificateIssuanceStatusChoices.ISSUED:
logger.warning(
"ChatBot %s is not ready. TLS certificate issuance status is %s",
self.rfc1034_compliant_name,
self.tls_certificate_issuance_status,
)
return False
if not self.deployed:
logger.warning("ChatBot %s is not ready. It is not deployed.", self.rfc1034_compliant_name)
return False
return True
[docs]
def mode(self, url: str) -> str:
"""
Determine the mode of the ChatBot based on the provided URL.
:param url: The URL to evaluate.
:returns: The mode of the ChatBot (sandbox, custom, default, unknown).
:rtype: str
"""
logger.debug("mode: %s", url)
if not url:
return self.Modes.UNKNOWN
SmarterValidator.validate_url(url)
url = SmarterValidator.urlify(url, environment=smarter_settings.environment) # type: ignore[return-value]
parsed_url = urlparse(url)
input_hostname = parsed_url.netloc
# most likely case first when running in production, at scale.
try:
default_url = SmarterValidator.urlify(self.default_host, environment=smarter_settings.environment) # type: ignore[return-value]
if default_url:
default_hostname = urlparse(default_url).netloc
if default_hostname and input_hostname == default_hostname:
return self.Modes.DEFAULT
except SmarterValueError:
pass
# workbench sandbox mode
try:
sandbox_url = SmarterValidator.urlify(self.sandbox_host, environment=smarter_settings.environment) # type: ignore[return-value]
if sandbox_url:
sandbox_hostname = urlparse(sandbox_url).netloc
if sandbox_hostname and input_hostname == sandbox_hostname:
return self.Modes.SANDBOX
except SmarterValueError:
pass
# custom domain mode. Least likely case.
try:
custom_url = SmarterValidator.urlify(self.custom_host, environment=smarter_settings.environment) # type: ignore[return-value]
if custom_url:
custom_hostname = urlparse(custom_url).netloc
if custom_hostname and input_hostname == custom_hostname:
return self.Modes.CUSTOM
except SmarterValueError:
pass
logger.error(
"Invalid ChatBot url %s received for default_url: %s, sandbox_url: %s, custom_url: %a",
url,
self.default_url,
self.sandbox_url,
self.custom_url,
)
# default to default mode as a safety measure
return self.Modes.UNKNOWN
[docs]
@classmethod
def get_cached_object(
cls,
invalidate: Optional[bool] = False,
pk: Optional[int] = None,
name: Optional[str] = None,
user: Optional[User] = None,
user_profile: Optional[UserProfile] = None,
account: Optional[Account] = None,
) -> "ChatBot":
"""
Retrieve a model instance using caching to optimize performance.
Example usage:
.. code-block:: python
# Retrieve a ChatBot instance by primary key with caching
chatbot = ChatBot.get_cached_object(pk=1)
# Retrieve a ChatBot instance by name and user profile with caching
chatbot = ChatBot.get_cached_object(name="example", user_profile=my_user_profile)
:param pk: The primary key of the model instance to retrieve.
:param name: The name of the model instance to retrieve.
:param user: The user associated with the model instance.
:param user_profile: The user profile associated with the model instance.
:param account: The account associated with the model instance.
:returns: The model instance if found, otherwise None.
:rtype: Optional["ChatBot"]
"""
logger_prefix = formatted_text(__name__ + "." + ChatBot.__name__ + ".get_cached_object()")
logger.debug(
"%s called %s with pk=%s, name=%s, user=%s, user_profile=%s, account=%s, invalidate=%s",
logger_prefix,
cls.__name__,
pk,
name,
user,
user_profile,
account,
invalidate,
)
retval = super().get_cached_object(invalidate=invalidate, pk=pk, name=name, user=user, user_profile=user_profile, account=account) # type: ignore[assignment]
if retval is None:
raise ChatBot.DoesNotExist(f"{cls.__name__} matching query does not exist.")
return retval # type: ignore[return-value]
[docs]
@classmethod
def get_cached_objects(
cls, invalidate: Optional[bool] = False, user_profile: Optional[UserProfile] = None
) -> models.QuerySet["ChatBot"]:
"""
Retrieve a list of ChatBot instances associated with a user profile using caching.
Example usage:
.. code-block:: python
# Retrieve ChatBot instances for a user profile with caching
chatbots = ChatBot.get_cached_objects(my_user_profile, invalidate=True)
:param invalidate: Whether to invalidate the cache for this retrieval.
:param user_profile: The user profile for which to retrieve ChatBot instances.
:returns: A queryset of ChatBot instances associated with the user profile.
:rtype: models.QuerySet["ChatBot"]
"""
logger_prefix = formatted_text(__name__ + "." + ChatBot.__name__ + ".get_cached_objects()")
logger.debug("%s called with user_profile=%s, invalidate=%s", logger_prefix, user_profile, invalidate)
@cache_results()
def _get_chatbots_for_user_profile_id(
user_profile_id: int, class_name: str = cls.__name__
) -> models.QuerySet["ChatBot"]:
return (
cls.objects.prefetch_related("tags")
.select_related("user_profile", "user_profile__account", "user_profile__user")
.filter(user_profile_id=user_profile_id)
)
if invalidate and user_profile:
_get_chatbots_for_user_profile_id.invalidate(user_profile_id=user_profile.id, class_name=cls.__name__) # type: ignore[union-attr]
if user_profile:
return _get_chatbots_for_user_profile_id(user_profile_id=user_profile.id, class_name=cls.__name__) # type: ignore[return-value]
return super().get_cached_objects(user_profile=user_profile, invalidate=invalidate) # type: ignore[return-value]
[docs]
def save(self, *args, asynchronous=False, **kwargs):
"""
Override save() to validate domain and send signals on status changes.
:raises SmarterValueError: If invalid hostname is provided.
:args: Positional arguments for the save method.
:asynchronous: If True, skips signal sending for asynchronous operations.
:kwargs: Keyword arguments for the save method.
:returns: None
"""
logger.debug("%s.save() called for ChatBot id: %s %s", self.formatted_class_name, self.pk, self.default_host)
if asynchronous:
logger.debug(
"%s.save() running in asynchronous mode for ChatBot id: %s. Skipping signal sending.",
self.formatted_class_name,
self.pk,
)
super().save(*args, **kwargs)
return
is_new = self.pk is None
SmarterValidator.validate_domain(self.hostname)
should_deploy = False
should_undeploy = False
if is_new:
if self.deployed:
chatbot_deploy.send(sender=self.__class__, chatbot=self)
else:
orig: ChatBot
try:
orig = ChatBot.objects.get(id=self.pk)
except ChatBot.DoesNotExist:
logger.error(
"%s.save() could not find original ChatBot with id: %s", self.formatted_class_name, self.pk
)
return super().save(*args, **kwargs)
if orig.dns_verification_status != self.dns_verification_status:
chatbot_dns_verification_status_changed.send(sender=self.__class__, chatbot=self)
chatbot_deploy_status_changed.send(sender=self.__class__, chatbot=self)
if self.dns_verification_status == ChatBot.DnsVerificationStatusChoices.VERIFYING:
chatbot_dns_verification_initiated.send(sender=self.__class__, chatbot=self)
if self.dns_verification_status == ChatBot.DnsVerificationStatusChoices.VERIFIED:
chatbot_dns_verified.send(sender=self.__class__, chatbot=self)
if self.dns_verification_status == ChatBot.DnsVerificationStatusChoices.FAILED:
chatbot_dns_failed.send(sender=self.__class__, chatbot=self)
chatbot_deploy_failed.send(sender=self.__class__, chatbot=self)
if self.deployed and not orig.deployed:
should_deploy = True
if not self.deployed and orig.deployed:
should_undeploy = True
super().save(*args, **kwargs)
if should_deploy:
logger.debug(
"%s.ChatBot.save() sending chatbot_deploy signal for ChatBot id: %s", self.formatted_class_name, self.pk
)
chatbot_deploy.send(sender=self.__class__, chatbot=self)
if should_undeploy:
logger.debug(
"%s.ChatBot.save() sending chatbot_undeploy signal for ChatBot id: %s",
self.formatted_class_name,
self.pk,
)
chatbot_undeploy.send(sender=self.__class__, chatbot=self)
[docs]
class ChatBotAPIKey(TimestampedModel):
"""
Represents the mapping of API keys to ChatBot instances within the Smarter platform.
.. important::
If present, the ChatBot associated with this record will require Api Key authentication
for all API requests. Otherwise, the ChatBot will allow anonymous unauthenticated access.
See :class:`smarter.lib.drf.token_authentication.SmarterTokenAuthentication` .
This model establishes a relationship between a ChatBot and its associated API keys,
enabling secure authentication and authorization for API access. Each entry in this
model links a specific ChatBot to a unique API key, allowing fine-grained control
over which keys can interact with which chatbot instances.
The ChatBotAPIKey model is essential for managing access to chatbot APIs, supporting
use cases such as per-bot API key rotation, revocation, and auditing. By associating
API keys with individual chatbots, the platform can enforce security policies and
monitor usage at the chatbot level.
Typical usage involves creating a ChatBotAPIKey instance whenever a new API key is
provisioned for a chatbot, and querying this model to validate incoming requests
against active keys.
**Model Relationships**
- Each ChatBotAPIKey is linked to one :class:`ChatBot` instance.
- Each ChatBotAPIKey references one :class:`SmarterAuthToken` representing the API key.
**Example**
.. code-block:: python
# Assign an API key to a chatbot
api_key = SmarterAuthToken.objects.create(...)
chatbot_api_key = ChatBotAPIKey.objects.create(chatbot=my_chatbot, api_key=api_key)
# Query API keys for a chatbot
keys = ChatBotAPIKey.objects.filter(chatbot=my_chatbot)
**Notes**
- API key activation and deactivation are managed via the SmarterAuthToken model.
- This model supports auditing and access control for chatbot API endpoints.
- Intended for internal use within the Smarter platform to secure chatbot APIs.
"""
class Meta:
verbose_name_plural = "ChatBot API Keys"
#: The ChatBot instance associated with this API key.
chatbot = models.ForeignKey(ChatBot, on_delete=models.CASCADE)
#: The API key (SmarterAuthToken) associated with the ChatBot.
api_key = models.ForeignKey(SmarterAuthToken, on_delete=models.CASCADE)
[docs]
@classmethod
def has_active_api_key(cls, chatbot: ChatBot, invalidate: Optional[bool] = False) -> bool:
"""
Returns True if the chatbot has at least one active API key.
"""
logger_prefix = formatted_text(__name__ + "." + cls.__name__ + ".has_active_api_key()")
logger.debug("%s called with chatbot=%s, invalidate=%s", logger_prefix, chatbot, invalidate)
@cache_results(cls.cache_expiration)
def _has_active_api_key(chatbot_id: int, class_name: str) -> bool:
return cls.objects.filter(chatbot_id=chatbot_id, api_key__is_active=True).exists()
if invalidate and chatbot:
_has_active_api_key.invalidate(chatbot_id=chatbot.id, class_name=ChatBotAPIKey.__name__) # type: ignore[union-attr]
if chatbot:
return _has_active_api_key(chatbot_id=chatbot.id, class_name=ChatBotAPIKey.__name__) # type: ignore[return-value]
return False
# pylint: disable=W0221
[docs]
@classmethod
def get_cached_objects(
cls, invalidate: Optional[bool] = False, chatbot: Optional[ChatBot] = None
) -> models.QuerySet["ChatBotAPIKey"]:
"""
Retrieve a list of ChatBotAPIKey instances associated with a ChatBot using caching.
Example usage:
.. code-block:: python
# Retrieve API keys for a chatbot with caching
api_keys = ChatBotAPIKey.get_cached_objects(my_chatbot, invalidate=True)
:param invalidate: Whether to invalidate the cache for this retrieval.
:type invalidate: bool, optional
:param chatbot: The ChatBot instance for which to retrieve API keys.
:type chatbot: ChatBot, optional
:returns: A queryset of ChatBotAPIKey instances associated with the ChatBot.
:rtype: models.QuerySet["ChatBotAPIKey"]
"""
logger_prefix = formatted_text(__name__ + "." + ChatBotAPIKey.__name__ + ".get_cached_objects()")
logger.debug("%s called with chatbot=%s, invalidate=%s", logger_prefix, chatbot, invalidate)
@cache_results(cls.cache_expiration)
def _get_api_keys_for_chatbot_id(
chatbot_id: int, class_name: str = cls.__name__
) -> models.QuerySet["ChatBotAPIKey"]:
return cls.objects.filter(chatbot_id=chatbot_id).select_related(
"chatbot",
"chatbot__user_profile",
"chatbot__user_profile__user",
"chatbot__user_profile__account",
"api_key",
"api_key__user_profile",
"api_key__user_profile",
"api_key__user_profile__user",
"api_key__user_profile__account",
)
if invalidate and chatbot:
_get_api_keys_for_chatbot_id.invalidate(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[union-attr]
if chatbot:
if ChatBotAPIKey.has_active_api_key(chatbot=chatbot, invalidate=invalidate):
return _get_api_keys_for_chatbot_id(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[return-value]
return ChatBotAPIKey.objects.none()
return super().get_cached_objects(invalidate=invalidate) # type: ignore[return-value]
[docs]
class ChatBotPlugin(TimestampedModel):
"""
Represents the association between a ChatBot instance and its enabled plugins within the Smarter platform.
This model establishes a many-to-one relationship, where each plugin entry is linked to a specific ChatBot
and references metadata describing the plugin. By maintaining this mapping, the platform can manage which
plugins are available to each chatbot, enabling extensibility and customization of chatbot capabilities.
The ChatBotPlugin model supports use cases such as plugin activation, deactivation, and enumeration for
individual chatbots. It is essential for scenarios where chatbots require additional functionality
provided by external or internal plugins, such as integrations, enhanced processing, or custom behaviors.
**Model Relationships**
- Each ChatBotPlugin is linked to one :class:`ChatBot` instance.
- Each ChatBotPlugin references one :class:`PluginMeta` instance, which contains metadata about the plugin.
**Usage Example**
.. code-block:: python
# Add a plugin to a chatbot
plugin_meta = PluginMeta.objects.get(name="weather")
chatbot_plugin = ChatBotPlugin.objects.create(chatbot=my_chatbot, plugin_meta=plugin_meta)
# List all plugins for a chatbot
plugins = ChatBotPlugin.objects.filter(chatbot=my_chatbot)
**Notes**
- Plugin management and loading are handled via the PluginController and related infrastructure.
- This model is intended for internal use to support dynamic extension of chatbot features.
- Uniqueness is enforced for each (chatbot, plugin_meta) pair to prevent duplicate plugin assignments.
"""
class Meta:
verbose_name_plural = "ChatBot Plugins"
unique_together = ("chatbot", "plugin_meta")
#: The ChatBot instance associated with this plugin.
chatbot = models.ForeignKey(ChatBot, on_delete=models.CASCADE)
#: The metadata for the plugin associated with the ChatBot.
plugin_meta = models.ForeignKey(PluginMeta, on_delete=models.CASCADE)
def __str__(self):
try:
url = self.chatbot.url if self.chatbot else "undefined chatbot"
plugin_name = self.plugin_meta.name if self.plugin_meta else "undefined plugin"
except ChatBot.DoesNotExist:
url = "undefined chatbot"
except PluginMeta.DoesNotExist:
plugin_name = "undefined plugin"
return f"{url} - {plugin_name}"
@property
def plugin(self) -> Optional[PluginBase]:
"""
Returns the Plugin instance associated with this ChatBotPlugin.
:returns: Plugin instance or None
:rtype: Optional[PluginBase]
"""
if not self.chatbot:
return None
admin_user = UserProfile.admin_for_account(self.chatbot.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("ChatBotPlugin.plugin() failed to find admin user for chatbot account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
@cache_results()
def get_cached_plugin_controller(
account_id: int,
user_id: int,
plugin_meta_id: int,
user_profile_id: int,
class_name: str = self.__class__.__name__,
) -> PluginController:
return PluginController(
account=self.chatbot.user_profile.cached_account,
user=admin_user,
plugin_meta=self.plugin_meta,
user_profile=user_profile,
)
plugin_controller = get_cached_plugin_controller(
account_id=self.chatbot.user_profile.cached_account.id,
user_id=admin_user.id, # type: ignore[union-attr]
plugin_meta_id=self.plugin_meta.id,
user_profile_id=user_profile.id, # type: ignore[union-attr]
class_name=self.__class__.__name__,
)
this_plugin = plugin_controller.plugin
return this_plugin
[docs]
@classmethod
def load(cls: Type["ChatBotPlugin"], chatbot: ChatBot, data) -> "ChatBotPlugin":
"""
Load (aka import) a plugin from a data file in yaml or json format.
:param chatbot: The ChatBot instance to associate with the plugin.
:param data: The plugin manifest data in yaml or json format.
:returns: The created ChatBotPlugin instance.
:rtype: ChatBotPlugin
See Also:
- :py:class:`smarter.apps.plugin.manifest.controller.PluginController`
- :py:class:`smarter.lib.manifest.loader.SAMLoader`
"""
if not chatbot:
return None
admin_user = UserProfile.admin_for_account(chatbot.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("ChatBotPlugin.plugin() failed to find admin user for chatbot account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
loader = SAMLoader(manifest=data)
manifest = SAMPluginCommon(**loader.json_data) # type: ignore[call-arg]
plugin_controller = PluginController(
account=chatbot.user_profile.cached_account, user=admin_user, user_profile=user_profile, manifest=manifest
)
plugin = plugin_controller.plugin
if not plugin or plugin.plugin_meta is None:
raise SmarterValueError("ChatBotPlugin.load() failed to load plugin from data file")
return cls.objects.create(chatbot=chatbot, plugin_meta=plugin.plugin_meta)
[docs]
@classmethod
def plugins(cls, chatbot: ChatBot) -> List[PluginBase]:
"""
Returns a list of Plugin instances associated with the given ChatBot.
:param chatbot: The ChatBot instance to retrieve plugins for.
:returns: List of Plugin instances.
:rtype: List[PluginBase]
:raises SmarterValueError: If admin user for chatbot account is not found
or if a plugin fails to load.
See Also:
- :py:class:`smarter.apps.plugin.controller.PluginController`
"""
if not chatbot:
return []
chatbot_plugins = cls.objects.filter(chatbot=chatbot)
admin_user = UserProfile.admin_for_account(chatbot.user_profile.cached_account)
if admin_user is None:
raise SmarterValueError("ChatBotPlugin.plugin() failed to find admin user for chatbot account")
user_profile = UserProfile.get_cached_object(invalidate=False, user=admin_user)
retval = []
for chatbot_plugin in chatbot_plugins:
plugin_controller = PluginController(
account=chatbot.user_profile.cached_account,
user=admin_user,
plugin_meta=chatbot_plugin.plugin_meta,
user_profile=user_profile,
)
if not plugin_controller or not plugin_controller.plugin:
raise SmarterValueError(
f"ChatBotPlugin.plugins() failed to load plugin for {chatbot_plugin.plugin_meta.name}"
)
retval.append(plugin_controller.plugin)
return retval
# pylint: disable=W0221
[docs]
@classmethod
def get_cached_objects(
cls, invalidate: Optional[bool] = False, chatbot: Optional[ChatBot] = None
) -> models.QuerySet["ChatBotPlugin"]:
"""
Retrieve a queryset of ChatBotPlugin instances associated with a ChatBot using caching.
:param invalidate: Whether to invalidate the cache for this retrieval.
:type invalidate: bool, optional
:param chatbot: The ChatBot instance for which to retrieve plugins.
:type chatbot: ChatBot, optional
:returns: A queryset of ChatBotPlugin instances associated with the ChatBot.
:rtype: models.QuerySet["ChatBotPlugin"]
"""
logger_prefix = formatted_text(__name__ + "." + ChatBotPlugin.__name__ + ".get_cached_objects()")
logger.debug("%s called with chatbot=%s, invalidate=%s", logger_prefix, chatbot, invalidate)
@cache_results()
def _get_plugins_for_chatbot_id(
chatbot_id: int, class_name: str = cls.__name__
) -> models.QuerySet["ChatBotPlugin"]:
"""
Caches the plugins for a chatbot by chatbot_id to optimize
performance and reduce database queries.
:param chatbot_id: The ID of the ChatBot for which to retrieve plugins.
:param class_name: The name of the class for cache key purposes.
:returns: A queryset of ChatBotPlugin instances associated with the ChatBot.
:rtype: models.QuerySet["ChatBotPlugin"]
"""
return cls.objects.filter(chatbot_id=chatbot_id).select_related(
"plugin_meta",
"plugin_meta__user_profile",
"plugin_meta__user_profile__user",
"plugin_meta__user_profile__account",
"chatbot__user_profile",
"chatbot__user_profile__user",
"chatbot__user_profile__account",
)
if invalidate and chatbot:
_get_plugins_for_chatbot_id.invalidate(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[union-attr]
if chatbot:
return _get_plugins_for_chatbot_id(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[return-value]
return super().get_cached_objects(invalidate=invalidate) # type: ignore[return-value]
[docs]
@classmethod
def plugins_json(cls, chatbot: ChatBot) -> List[dict]:
retval = []
for plugin in cls.plugins(chatbot):
retval.append(plugin.to_json())
return retval
[docs]
class ChatBotFunctions(TimestampedModel):
"""
Represents the set of callable functions that are available to a ChatBot instance within the Smarter platform.
This model is used to define and manage the specific functions that a chatbot can access or invoke during its operation.
Each record in this model links a chatbot to a named function, enabling fine-grained control over the chatbot's capabilities.
The available functions are defined by a fixed set of choices, such as "weather", "news", "prices", and "math".
By associating functions with chatbots, the platform allows for extensible and customizable chatbot behavior, supporting
use cases where different chatbots require access to different sets of features or integrations. This model is essential
for scenarios where chatbots need to perform actions, retrieve information, or interact with external APIs in a controlled
and auditable manner.
**Model Relationships**
- Each ChatBotFunctions entry is linked to one :class:`ChatBot` instance.
- Each entry specifies a function name from a predefined set of choices.
**Usage Example**
.. code-block:: python
# Assign a function to a chatbot
ChatBotFunctions.objects.create(chatbot=my_chatbot, name="weather")
# List all functions available to a chatbot
functions = ChatBotFunctions.objects.filter(chatbot=my_chatbot)
**Notes**
- The set of available functions is controlled by the ``CHOICES`` class attribute.
- This model is intended for internal use to manage and audit chatbot capabilities.
- Uniqueness is not enforced, so a chatbot may have multiple entries for the same function if needed.
"""
class Meta:
verbose_name_plural = "ChatBot Functions"
CHOICES = [
("get_current_weather", "get_current_weather"),
("date_calculator", "date_calculator"),
("calculator", "calculator"),
]
"""
The set of available function names that can be assigned to a ChatBot.
See Also:
- :func:`smarter.apps.prompt.functions.function_weather.get_current_weather`
- :func:`smarter.apps.prompt.functions.function_date_calculator.date_calculator`
- :func:`smarter.apps.prompt.functions.function_calculator.calculator`
"""
#: The ChatBot instance associated with this function.
#: Example: ChatBot(id=1, name="my-chatbot")
chatbot = models.ForeignKey(ChatBot, on_delete=models.CASCADE)
#: The name of the function available to the ChatBot.
#: Example: "weather"
name = models.CharField(max_length=255, choices=CHOICES, blank=True, null=True)
[docs]
@classmethod
def choices_list(cls):
return [item[0] for item in cls.CHOICES]
[docs]
@classmethod
def functions(cls, chatbot: ChatBot) -> List[str]:
"""
Returns a list of function names associated with the given ChatBot.
:param chatbot: The ChatBot instance to retrieve functions for.
:returns: List of function names.
:rtype: List[str]
"""
if not chatbot:
return []
chatbot_functions = cls.objects.filter(chatbot=chatbot)
retval = [chatbot_function.name for chatbot_function in chatbot_functions if chatbot_function.name]
return retval
# pylint: disable=W0221
[docs]
@classmethod
def get_cached_objects(
cls, invalidate: Optional[bool] = False, chatbot: Optional[ChatBot] = None
) -> models.QuerySet["ChatBotFunctions"]:
"""
Retrieve a queryset of ChatBotFunctions instances associated with a ChatBot using caching.
:param invalidate: Whether to invalidate the cache for this retrieval.
:type invalidate: bool, optional
:param chatbot: The ChatBot instance for which to retrieve functions.
:type chatbot: ChatBot, optional
:returns: A queryset of ChatBotFunctions instances associated with the ChatBot.
:rtype: models.QuerySet["ChatBotFunctions"]
"""
logger_prefix = formatted_text(__name__ + "." + ChatBotFunctions.__name__ + ".get_cached_objects()")
logger.debug("%s called with chatbot=%s, invalidate=%s", logger_prefix, chatbot, invalidate)
@cache_results(cls.cache_expiration)
def _get_functions_for_chatbot_id(
chatbot_id: int, class_name: str = cls.__name__
) -> models.QuerySet["ChatBotFunctions"]:
"""
Caches the functions for a chatbot by chatbot_id to optimize
performance and reduce database queries.
:param chatbot_id: The ID of the ChatBot for which to retrieve functions.
:param class_name: The name of the class for cache key purposes.
:returns: A queryset of ChatBotFunctions instances associated with the ChatBot.
:rtype: models.QuerySet["ChatBotFunctions"]
"""
return cls.objects.filter(chatbot_id=chatbot_id).select_related(
"plugin_meta",
"plugin_meta__user_profile",
"plugin_meta__user_profile__user",
"plugin_meta__user_profile__account",
"chatbot__user_profile",
"chatbot__user_profile__user",
"chatbot__user_profile__account",
)
if invalidate and chatbot:
_get_functions_for_chatbot_id.invalidate(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[union-attr]
if chatbot:
return _get_functions_for_chatbot_id(chatbot_id=chatbot.id, class_name=cls.__name__) # type: ignore[return-value]
return super().get_cached_objects(invalidate=invalidate) # type: ignore[return-value]
[docs]
class ChatBotRequests(TimestampedModel):
"""
Stores the request history for a ChatBot instance within the Smarter platform.
This model is designed to record and manage all incoming requests made to a chatbot, providing a persistent audit trail
of interactions for analysis, debugging, and reporting. Each record in this model captures the details of a single request,
including the associated chatbot, the request payload, session information, and aggregation status.
**Purpose and Usage**
The ChatBotRequests model enables comprehensive tracking of chatbot usage and user interactions. By storing each request,
the platform can support features such as:
- Request analytics and reporting for chatbot performance and user engagement.
- Debugging and trouble shooting of chatbot behavior by reviewing historical requests.
- Session management, allowing grouping and correlation of requests within a user session.
- Aggregation of requests for batch processing or summarization.
**Model Relationships**
- Each ChatBotRequests entry is linked to one :class:`ChatBot` instance, establishing a direct association between the request and the chatbot that handled it.
**Notes**
- This model is intended for internal use to support auditing, analytics, and operational monitoring of chatbot activity.
- The request data is stored in JSON format to accommodate flexible and extensible payload structures.
- Aggregation support allows for efficient handling of bulk or grouped requests, which may be relevant for advanced chatbot workflows.
**Example Usage**
.. code-block:: python
# Record a new request for a chatbot
ChatBotRequests.objects.create(
chatbot=my_chatbot,
request={"message": "Hello, chatbot!"},
session_key="abc123",
is_aggregation=False,
)
# Retrieve all requests for a specific chatbot
requests = ChatBotRequests.objects.filter(chatbot=my_chatbot)
See Also:
- :mod:`smarter.apps.chatbot.tasks`
"""
class Meta:
verbose_name_plural = "ChatBot Requests History"
chatbot = models.ForeignKey(ChatBot, on_delete=models.CASCADE)
request = models.JSONField(
blank=True,
null=True,
encoder=json.SmarterJSONEncoder,
)
session_key = models.CharField(max_length=255, blank=True, null=True)
is_aggregation = models.BooleanField(default=False, blank=True, null=True)
class ChatBotRequestsSerializer(serializers.ModelSerializer):
class Meta:
model = ChatBotRequests
fields = (
"id",
"created_at",
"updated_at",
"request",
"is_aggregation",
)
class ChatBotSerializer(serializers.ModelSerializer):
url_chatbot = serializers.ReadOnlyField()
user_profile = UserProfileSerializer()
class Meta:
model = ChatBot
fields = []
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.Meta.fields = [
field.name
for field in self.Meta.model._meta.get_fields()
if field.name not in ["chat", "chatbotapikey", "chatbotplugin", "chatbotfunctions", "chatbotrequests"]
]
self.Meta.fields += ["url_chatbot", "user_profile"]
class ChatBotCustomDomainSerializer(serializers.ModelSerializer):
class Meta:
model = ChatBotCustomDomain
fields = "__all__"
def get_cached_chatbot_by_request(request: HttpRequest) -> Optional[ChatBot]:
"""
Returns the chatbot from the cache if it exists, otherwise
it queries the database with assistance from ChatBotHelper
and caches the result.
.. code-block:: python
chatbot = get_cached_chatbot_by_request(request)
print(chatbot.url)
param request: The Django HttpRequest object containing the URL and user context.
type request: django.http.HttpRequest
returns: The ChatBot instance associated with the request URL, or None if not found.
rtype: Optional[ChatBot]
"""
@cache_results()
def get_chatbot_by_url(url: str, class_name: str) -> Optional[ChatBot]:
"""
We use the request URL as the cache key to avoid redundant
parsing and database queries for repeated requests.
"""
chatbot_helper = ChatBotHelper(request)
if chatbot_helper:
logger.debug(
"%s.get_cached_chatbot_by_request() resolved and cached chatbot '%s' for url: %s",
formatted_text(__name__),
chatbot_helper.chatbot,
url,
)
return chatbot_helper.chatbot
if not request:
return None
url = smarter_build_absolute_uri(request)
return get_chatbot_by_url(url=url, class_name=ChatBot.__name__)
[docs]
class ChatBotHelper(SmarterRequestMixin):
"""
Provides a mapping between URLs and their corresponding ChatBot models,
abstracting URL parsing logic for reuse across the codebase.
This helper class is designed to centralize and standardize the logic
required to resolve a ChatBot instance from a given URL or request context.
It is intended for use in various locations, including within this module,
Django middleware, and view logic.
The class also implements caching of ChatBot objects for specific URLs,
reducing redundant parsing and database queries for repeated requests.
**Supported URL Patterns**
The following are examples of valid URLs that this helper can process:
- **Authentication Optional URLs:**
- ``https://example-username.3141-5926-5359.alpha.api.example.com/``
- ``https://example-username.3141-5926-5359.alpha.api.example.com/config/``
- **Authenticated URLs:**
- ``https://alpha.api.example-username.com/smarter/example/``
- ``https://example-username.smarter.sh/chatbot/``
- ``https://alpha.api.example-username.com/workbench/1/``
- ``https://alpha.api.example-username.com/workbench/example/``
- **Legacy (pre v0.12) URLs:**
- ``https://alpha.api.example-username.com/chatbots/1/``
- ``https://alpha.api.example-username.com/chatbots/example/``
where for ``example-username``, ``example`` is the ChatBot name,
``username`` is the Account Username, and ``3141-5926-5359`` is the
Account Number.
**Features**
- Abstracts and encapsulates URL parsing and ChatBot resolution logic.
- Provides a consistent interface for retrieving ChatBot instances from URLs.
- Caches ChatBot objects to avoid redundant lookups.
- Supports both authenticated and unauthenticated URL patterns.
- Handles legacy URL formats for backward compatibility.
**Usage**
This class is typically instantiated with a Django ``HttpRequest`` object.
It can then be used to access the resolved ChatBot instance and related
metadata, such as the associated account, chatbot ID, and custom domain.
Example::
helper = ChatBotHelper(request)
chatbot = helper.chatbot
if helper.is_valid:
# Proceed with chatbot logic
:param request: The Django HttpRequest object containing the URL and user context.
:type request: django.http.HttpRequest
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments, such as 'chatbot', 'chatbot_custom_domain', etc.
:raises SmarterConfigurationError: If the helper cannot resolve a valid ChatBot instance.
.. note::
This class is intended for internal use within the Smarter platform and
should not be used directly in user-facing code without proper validation.
"""
__slots__ = (
"_chatbot",
"_chatbot_custom_domain",
"_chatbot_requests",
"_chatbot_id",
"_name",
"_is_chatbothelper_ready",
)
[docs]
def __init__(self, request: HttpRequest, *args, **kwargs):
"""
Initializes the ChatBotHelper instance.
:param request: The Django HttpRequest object.
:type request: django.http.HttpRequest
:param args: Additional positional arguments.
:param kwargs: Additional keyword arguments.
"""
self._chatbot = None
self._chatbot_custom_domain = None
self._chatbot_requests = None
self._chatbot_id = None
self._name = None
self._is_chatbothelper_ready: bool = False
chatbot_helper_logger.debug(
"%s.__init__() called with url: %s args: %s, kwargs: %s",
self.formatted_class_name,
request.build_absolute_uri() if request else None,
args,
kwargs,
)
self._chatbot: Optional[ChatBot] = kwargs.get("chatbot")
if isinstance(self._chatbot, ChatBot):
chatbot_helper_logger.debug(
"%s.__init__() received ChatBot: %s",
self.formatted_class_name,
str(self._chatbot),
)
self._chatbot_id: Optional[int] = kwargs.get("chatbot_id")
if isinstance(self._chatbot_id, int):
chatbot_helper_logger.debug(
"%s.__init__() received chatbot_id: %s",
self.formatted_class_name,
str(self._chatbot_id),
)
self._name: Optional[str] = kwargs.get("name")
if isinstance(self._name, str):
chatbot_helper_logger.debug(
"%s.__init__() received name: %s",
self.formatted_class_name,
str(self._name),
)
self._chatbot_custom_domain: Optional[ChatBotCustomDomain] = kwargs.get("chatbot_custom_domain")
if isinstance(self._chatbot_custom_domain, ChatBotCustomDomain):
chatbot_helper_logger.debug(
"%s.__init__() received ChatBotCustomDomain: %s",
self.formatted_class_name,
str(self._chatbot_custom_domain),
)
self._chatbot_requests: Optional[ChatBotRequests] = kwargs.get("chatbot_requests")
if isinstance(self._chatbot_requests, ChatBotRequests):
chatbot_helper_logger.debug(
"%s.__init__() received ChatBotRequests: %s",
self.formatted_class_name,
str(self._chatbot_requests),
)
# initializations that depend on the superclass
super().__init__(request, *args, **kwargs)
chatbot_helper_logger.debug("%s.__init__() completed super().__init__()", self.formatted_class_name)
self._chatbot_id = self._chatbot_id or self.smarter_request_chatbot_id
self._name = self._name or self.smarter_request_chatbot_name
if self.is_chatbot:
if not isinstance(self.chatbot, ChatBot):
if self.user_profile and self._name:
try:
self.chatbot = ChatBot.get_cached_object(name=self._name, user_profile=self.user_profile)
except ChatBot.DoesNotExist:
chatbot_helper_logger.warning(
"%s.__init__() could not find ChatBot with name=%s and user_profile=%s",
self.formatted_class_name,
self._name,
self.user_profile,
)
if not isinstance(self._chatbot, ChatBot):
chatbot_helper_logger.warning(
"%s.__init__() did not find a ChatBot for url=%s, name=%s, chatbot_id=%s, user_profile=%s",
self.formatted_class_name,
self.url,
self.name,
self.chatbot_id,
self.user_profile,
)
msg = f"{self.formatted_class_name}.__init__() is {self.chatbothelper_ready_state} - {self.chatbot if self.chatbot else 'ChatBot not initialized'} - {self.user_profile if self.user_profile else 'UserProfile not initialized'}"
if self.ready:
chatbot_helper_logger.debug(msg)
chatbot_helper_logger.debug(
"%s.__init__() initialized with url=%s, name=%s, chatbot_id=%s, user=%s, user_profile=%s, session_key=%s",
self.formatted_class_name,
self.url if self.url else "undefined",
self.name,
self.chatbot_id,
self.user,
self.user_profile,
self.session_key,
)
else:
chatbot_helper_logger.error(msg)
def __str__(self):
return str(self.chatbot) if self._chatbot else "undefined"
[docs]
@cached_property
def account(self) -> Optional[Account]:
"""
Return the associated :class:`Account` for this ChatBotHelper instance,
optionally overriding the default account based on the account number
parsed from the URL, if available.
If the URL contains an account number (for example,
``http://education.3141-5926-5359.api.localhost:9357/config/``),
this method will attempt to retrieve and return the corresponding
cached Account object. If no account number is found in the URL,
the default account from the superclass is returned.
:returns: The resolved :class:`Account` instance, or ``None`` if not found.
:rtype: Optional[Account]
"""
account_number = account_number_from_url(self._url) # type: ignore[arg-type]
if account_number:
chatbot_helper_logger.debug("overriding account with account_number from named url: %s", self.url)
return Account.get_cached_object(account_number=account_number)
# from the super()
return self._account
@property
def chatbot_id(self) -> Optional[int]:
"""
Returns the :attr:`ChatBot.id` for this ChatBotHelper instance.
This property attempts to resolve the ChatBot's unique integer ID using several strategies:
1. If a chatbot ID was provided at initialization, it is returned immediately.
2. If a ChatBot object is already cached, its ID is returned.
3. If the parent :class:`SmarterRequestMixin` provides a chatbot ID (e.g., parsed from the URL), it is used.
4. If both a chatbot name and account are available, attempts to resolve and cache the ChatBot object and its ID.
:returns: The resolved ChatBot ID, or ``None`` if not found.
:rtype: Optional[int]
"""
# check for a value passed in
if self._chatbot_id:
return self._chatbot_id
# check for a chatbot object
if self._chatbot:
self._chatbot_id = self.chatbot.id # type: ignore[return-value]
return self._chatbot_id
# check SmarterRequestMixin for a chatbot_id derived from the url
self._chatbot_id = super().smarter_request_chatbot_id
if self._chatbot_id:
return self._chatbot_id
if self.chatbot_name and self.user_profile:
self.chatbot = ChatBot.get_cached_object(name=self.chatbot_name, user_profile=self.user_profile)
chatbot_helper_logger.debug(
f"chatbot_id() initialized self.chatbot_id={self.chatbot_id} from name={ self.chatbot_name } and account={ self.account }"
)
return self._chatbot_id
return self._chatbot_id
@chatbot_id.setter
def chatbot_id(self, chatbot_id: int):
self._chatbot_id = chatbot_id
chatbot = ChatBot.get_cached_object(pk=chatbot_id)
if chatbot and chatbot.user_profile.cached_account != self.account:
raise SmarterValueError("ChatBotHelper.chatbot_id setter: ChatBot's Account does not match self.account")
self.chatbot = chatbot
if self._chatbot:
chatbot_helper_logger.debug(
f"@chatbot_id.setter initialized self.chatbot_id={self.chatbot_id} from chatbot_id"
)
@property
def chatbot_name(self) -> Optional[str]:
"""
Returns the ChatBot.name for the ChatBotHelper.
"""
return self.name
@property
def name(self) -> Optional[str]:
"""
Returns the name of the chatbot.
This property attempts to resolve the chatbot's name using several strategies, in order of precedence:
1. ``self._name``: The name assigned during initialization, if available.
2. ``self.chatbot.name``: The name attribute of the resolved ChatBot instance, if present.
3. ``self.subdomain``: If the URL is a named chatbot URL (i.e., ``is_chatbot_named_url`` is True), the subdomain is used as the name.
4. Path slug: If the URL is a sandbox chatbot URL (i.e., ``is_chatbot_sandbox_url`` is True), the path slug is used as the name.
:returns: The resolved chatbot name, or ``None`` if not found.
:rtype: Optional[str]
"""
if self._chatbot:
self._name = self._chatbot.name
return self._name
@property
def rfc1034_compliant_name(self) -> Optional[str]:
"""
Returns a URL-friendly name for the chatbot.
This is a convenience property that returns a RFC 1034 compliant name for the chatbot.
Examples
--------
.. code-block:: python
self.name # 'Example ChatBot 1'
self.rfc1034_compliant_name # 'example-chatbot-1'
:returns: The RFC 1034 compliant name for the chatbot, or ``None`` if not available.
:rtype: Optional[str]
"""
if self._chatbot:
return self._chatbot.rfc1034_compliant_name
return None
[docs]
@cached_property
def is_chatbothelper_ready(self) -> bool:
"""
Returns ``True`` if the ChatBotHelper is ready to be used.
This is a convenience property that checks if the ChatBotHelper
is initialized and has a valid :class:`ChatBot` instance.
:returns: ``True`` if the helper is initialized and has a valid ChatBot, otherwise ``False``.
:rtype: bool
"""
if self._is_chatbothelper_ready:
return True
logger_prefix = f"{self.formatted_class_name}.is_chatbothelper_ready()"
if isinstance(self._chatbot, ChatBot):
chatbot_helper_logger.debug(
"%s returning true because chatbot is initialized: %s",
logger_prefix,
self._chatbot,
)
self._is_chatbothelper_ready = True
return self._is_chatbothelper_ready
if self.chatbot_custom_domain:
chatbot_helper_logger.debug(
"%s chatbot_custom_domain is set but ChatBotHelpler is not confirmed to be ready.",
logger_prefix,
)
if not self.is_chatbot:
chatbot_helper_logger.debug(
"%s returning false because is_chatbot is false",
logger_prefix,
)
return False
else:
chatbot_helper_logger.debug(
"%s confirmed URL is a chatbot URL. url=%s",
logger_prefix,
self._url,
)
if not self.user or not self.user.is_authenticated:
chatbot_helper_logger.warning(
"%s returning false because called with unauthenticated request",
logger_prefix,
)
return False
else:
chatbot_helper_logger.debug(
"%s confirmed request user is authenticated: %s",
logger_prefix,
self.user.username,
)
if not self.account:
chatbot_helper_logger.warning("%s returning false because called with no account", logger_prefix)
return False
else:
chatbot_helper_logger.debug(
"%s confirmed account is assigned: %s",
logger_prefix,
self.account,
)
if not isinstance(self.name, str):
chatbot_helper_logger.warning(
"%s returning false because did not find a name for the chatbot.", logger_prefix
)
return False
else:
chatbot_helper_logger.debug(
"%s confirmed chatbot name is assigned: %s",
logger_prefix,
self.name,
)
if not isinstance(self._chatbot, ChatBot):
chatbot_helper_logger.debug(
"%s returning false because ChatBot is not initialized.",
logger_prefix,
)
return False
else:
chatbot_helper_logger.debug(
"%s confirmed ChatBot is initialized: %s",
logger_prefix,
self._chatbot,
)
self._is_chatbothelper_ready = True
return self._is_chatbothelper_ready
@property
def chatbothelper_ready_state(self) -> str:
"""
Returns a formatted string indicating whether the ChatBotHelper is ready.
:return: A string indicating whether the ChatBotHelper is ready or not.
"""
return formatted_text_green("Ready") if self.is_chatbothelper_ready else formatted_text_red("Not Ready")
@property
def ready(self) -> bool:
"""
Returns ``True`` if the ChatBotHelper and its ChatBot are ready to be used.
This property checks both the readiness of the ChatBotHelper itself and the readiness
of the underlying ChatBot instance.
:returns: ``True`` if both the helper and ChatBot are ready, otherwise ``False``.
:rtype: bool
"""
# there is a scenario where the SmarterRequestMixin is not ready but the ChatBotHelper is.
if self.is_chatbothelper_ready and self.user_profile and not super().ready:
chatbot_helper_logger.debug(
"%s.ready() returning true because ChatBotHelper is ready even though SmarterRequestMixin is not ready",
self.formatted_class_name,
)
return True
if not super().ready:
chatbot_helper_logger.debug(
"%s.ready() returning false because SmarterRequestMixin is not ready", self.formatted_class_name
)
return False
return self.is_chatbothelper_ready
[docs]
def to_json(self) -> dict[str, Any]:
"""
Serialize the ChatBotHelper to a dictionary.
This method returns a dictionary representation of the ChatBotHelper instance,
including key metadata and related objects such as the chatbot, account, and custom domain.
:returns: A dictionary containing the serialized state of the ChatBotHelper.
:rtype: dict[str, Any]
"""
return self.sorted_dict(
{
"ready": self.ready,
"name": self.name,
"api_host": self.api_host,
"chatbot_id": self.chatbot_id,
"chatbot_name": self.chatbot_name,
"chatbot_custom_domain": (
ChatBotCustomDomainSerializer(self.chatbot_custom_domain) if self.chatbot_custom_domain else None
),
"environment_api_domain": smarter_settings.environment_api_domain,
"is_custom_domain": self.is_custom_domain,
"is_deployed": self.is_deployed,
"is_authentication_required": self.is_authentication_required,
"is_chatbothelper_ready": self.is_chatbothelper_ready,
"rfc1034_compliant_name": self.rfc1034_compliant_name,
"chatbot": ChatBotSerializer(self.chatbot).data if self.chatbot else None,
"url": self.url,
**super().to_json(),
}
)
[docs]
@cached_property
def api_host(self) -> Optional[str]:
"""
Returns the API host for a ChatBot API URL.
This property extracts and returns the API host component from the chatbot URL,
supporting named, sandbox, and custom domain URLs.
Examples
--------
Named URL:
- ``https://hr.3141-5926-5359.alpha.api.example.com/chatbot/``
returns ``'alpha.api.example.com'``
Sandbox URL:
- ``http://api.localhost:9357/api/v1/chatbots/1/chat/``
returns ``'api.localhost:9357'``
Custom domain URL:
- ``https://hr.smarter.sh/chatbot/``
returns ``'hr.smarter.sh'``
:returns: The API host as a string, or ``None`` if not found.
:rtype: Optional[str]
"""
if not self.smarter_request:
return None
if not self.qualified_request:
return None
if self.is_smarter_api and isinstance(self._url, ParseResult):
return self._url.netloc
if self.is_custom_domain and isinstance(self._url, ParseResult):
# example: hr.bots.example.com
return self._url.netloc
return smarter_settings.environment_api_domain
@property
def is_deployed(self) -> bool:
return self.chatbot.deployed if self.chatbot else False # type: ignore[return-value]
[docs]
@cached_property
def is_authentication_required(self) -> bool:
"""
Determines if authentication is required to access the ChatBot.
:returns: ``True`` if authentication is required, otherwise ``False``.
:rtype: bool
"""
if self.is_chatbot_sandbox_url:
return True
if not self.chatbot:
return False
chatbotapikeys = ChatBotAPIKey.get_cached_objects(chatbot=self.chatbot)
if chatbotapikeys.filter(api_key__is_active=True).exists():
return True
return False
@property
def chatbot(self) -> Optional[ChatBot]:
"""
Returns a lazy instance of the ChatBot.
Examples
--------
- https://hr.3141-5926-5359.alpha.api.example.com/chatbot/
returns ChatBot(name='hr', account=Account(...))
:returns: The ChatBot instance, or ``None`` if not found.
:rtype: Optional[ChatBot]
"""
if self._chatbot:
return self._chatbot
# cheapest possibility
if self._chatbot_id:
self.chatbot = ChatBot.get_cached_object(pk=self._chatbot_id)
chatbot_helper_logger.debug(f"initialized chatbot {self._chatbot} from chatbot_id {self.chatbot_id}")
return self._chatbot
# our expected case
if self.user_profile and self.name:
try:
self.chatbot = ChatBot.get_cached_object(name=self.name, user_profile=self.user_profile)
chatbot_helper_logger.debug(
f"initialized chatbot {self._chatbot} from account {self.account} and name {self.name}"
)
return self._chatbot
except ChatBot.DoesNotExist:
chatbot_helper_logger.error(
"%s.chatbot() did not find chatbot for %s name: %s",
self.formatted_class_name,
self._user_profile,
self.name,
)
return self._chatbot
@chatbot.setter
def chatbot(self, chatbot: ChatBot):
"""
Sets the ChatBot instance for this ChatBotHelper.
"""
self._chatbot = chatbot
if self._chatbot:
self._chatbot_id = self._chatbot.id # type: ignore[assignment]
self._name = self._chatbot.name
chatbot_helper_logger.debug(
f"@chatbot.setter initialized self.chatbot_id={self.chatbot_id} and self.name={self.name} from chatbot"
)
else:
self._chatbot_id = None
self._name = None
chatbot_helper_logger.debug("@chatbot.setter cleared self.chatbot_id and self.name because chatbot is None")
if hasattr(self, "is_chatbothelper_ready"):
del self.is_chatbothelper_ready
[docs]
@cached_property
def provider(self) -> Optional[Provider]:
"""
Returns the Provider associated with the ChatBot.
:returns: The Provider instance, or ``None`` if not found.
:rtype: Optional[Provider]
"""
if not self.chatbot:
return None
try:
# FIX NOTE: self.chatbot.provider should be a foreign key to Provider.
return Provider.get_cached_object(name=self.chatbot.provider, account=self.account) # type: ignore[return-value]
except Provider.DoesNotExist:
return None
@property
def chatbot_plugins_list(self) -> list[ChatBotPlugin]:
"""
Returns a list of ChatBotPlugin instances associated with the ChatBot.
:returns: A list of ChatBotPlugin instances.
:rtype: list[ChatBotPlugin]
"""
if not self.chatbot:
return []
return list(ChatBotPlugin.get_cached_objects(chatbot=self.chatbot))
[docs]
@cached_property
def chatbot_plugins_list_str(self) -> str:
"""
Returns a comma-separated string of ChatBotPlugin names associated with the ChatBot.
:returns: A comma-separated string of ChatBotPlugin names.
:rtype: str
"""
plugins = self.chatbot_plugins_list
return ", ".join(
str(plugin.plugin_meta.name) + " (" + str(plugin.plugin_meta.user_profile) + ")" for plugin in plugins
)
@property
def is_custom_domain(self) -> bool:
"""
Returns ``True`` if the ChatBot is using a custom domain.
:returns: ``True`` if a custom domain is configured, otherwise ``False``.
:rtype: bool
"""
return self.chatbot_custom_domain is not None
@property
def chatbot_custom_domain(self) -> Optional[ChatBotCustomDomain]:
"""
Returns a lazy instance of the ChatBotCustomDomain.
Examples
--------
- ``https://hr.smarter.sh/chatbot/``
returns ``ChatBotCustomDomain(domain_name='smarter.sh')``
:returns: The ChatBotCustomDomain instance, or ``None`` if not found.
:rtype: Optional[ChatBotCustomDomain]
"""
if self._chatbot_custom_domain:
return self._chatbot_custom_domain
try:
self._chatbot_custom_domain = ChatBotCustomDomain.objects.get(
user_profile=self.user_profile, domain_name=self.root_domain
)
logger.debug(
"%s.chatbot_custom_domain() found ChatBotCustomDomain for root domain: %s %s",
self.formatted_class_name,
self.root_domain,
self.user_profile,
)
except ChatBotCustomDomain.DoesNotExist:
if not self.account:
logger.warning(
"%s.chatbot_custom_domain() cannot lookup ChatBotCustomDomain for rootdomain: %s because account is None",
self.formatted_class_name,
self.root_domain,
)
return None
account_admin = get_cached_admin_user_for_account(account=self.account) # type: ignore[arg-type]
account_admin_user_profile = UserProfile.get_cached_object(user=account_admin) # type: ignore[arg-type]
try:
self._chatbot_custom_domain = ChatBotCustomDomain.objects.get(
user_profile=account_admin_user_profile,
domain_name=self.root_domain,
)
logger.debug(
"%s.chatbot_custom_domain() found ChatBotCustomDomain for rootdomain: %s under account admin user_profile id %s",
self.formatted_class_name,
self.root_domain,
account_admin_user_profile,
)
except ChatBotCustomDomain.DoesNotExist:
try:
self._chatbot_custom_domain = ChatBotCustomDomain.objects.get(
user_profile=smarter_cached_objects.smarter_admin_user_profile,
domain_name=self.root_domain,
)
logger.debug(
"%s.chatbot_custom_domain() found ChatBotCustomDomain for rootdomain: %s under smarter platform admin user_profile id %s",
self.formatted_class_name,
self.root_domain,
smarter_cached_objects.smarter_admin_user_profile,
)
except ChatBotCustomDomain.DoesNotExist:
pass
if not self._chatbot_custom_domain:
logger.debug(
"%s.chatbot_custom_domain() did not find ChatBotCustomDomain for rootdomain: %s",
self.formatted_class_name,
self.root_domain,
)
return self._chatbot_custom_domain