text: import logging
  from datetime import timedelta
  from urllib.parse import parse_qsl, urlparse
  
  from django.apps import apps
  from django.conf import settings
  from django.core.exceptions import ImproperlyConfigured
  from django.db import models, transaction
  from django.urls import reverse
  from django.utils import timezone
  from django.utils.translation import gettext_lazy as _
  
  from .generators import generate_client_id, generate_client_secret
  from .scopes import get_scopes_backend
  from .settings import oauth2_settings
  from .validators import RedirectURIValidator, WildcardSet
  
  
  logger = logging.getLogger(__name__)
  
  
  class AbstractApplication(models.Model):
      """
      An Application instance represents a Client on the Authorization server.
      Usually an Application is created manually by client's developers after
      logging in on an Authorization Server.
  
      Fields:
  
      * :attr:`client_id` The client identifier issued to the client during the
                          registration process as described in :rfc:`2.2`
      * :attr:`user` ref to a Django user
      * :attr:`redirect_uris` The list of allowed redirect uri. The string
                              consists of valid URLs separated by space
      * :attr:`client_type` Client type as described in :rfc:`2.1`
      * :attr:`authorization_grant_type` Authorization flows available to the
                                         Application
      * :attr:`client_secret` Confidential secret issued to the client during
                              the registration process as described in :rfc:`2.2`
      * :attr:`name` Friendly name for the Application
      """
      CLIENT_CONFIDENTIAL = "confidential"
      CLIENT_PUBLIC = "public"
      CLIENT_TYPES = (
          (CLIENT_CONFIDENTIAL, _("Confidential")),
          (CLIENT_PUBLIC, _("Public")),
      )
  
      GRANT_AUTHORIZATION_CODE = "authorization-code"
      GRANT_IMPLICIT = "implicit"
      GRANT_PASSWORD = "password"
      GRANT_CLIENT_CREDENTIALS = "client-credentials"
      GRANT_TYPES = (
          (GRANT_AUTHORIZATION_CODE, _("Authorization code")),
          (GRANT_IMPLICIT, _("Implicit")),
          (GRANT_PASSWORD, _("Resource owner password-based")),
          (GRANT_CLIENT_CREDENTIALS, _("Client credentials")),
      )
  
      id = models.BigAutoField(primary_key=True)
      client_id = models.CharField(
          max_length=100, unique=True, default=generate_client_id, db_index=True
      )
      user = models.ForeignKey(
          settings.AUTH_USER_MODEL,
          related_name="%(app_label)s_%(class)s",
          null=True, blank=True, on_delete=models.CASCADE
      )
  
      redirect_uris = models.TextField(
          blank=True, help_text=_("Allowed URIs list, space separated"),
      )
      client_type = models.CharField(max_length=32, choices=CLIENT_TYPES)
      authorization_grant_type = models.CharField(
          max_length=32, choices=GRANT_TYPES
      )
      client_secret = models.CharField(
          max_length=255, blank=True, default=generate_client_secret, db_index=True
      )
      name = models.CharField(max_length=255, blank=True)
      skip_authorization = models.BooleanField(default=False)
  
      created = models.DateTimeField(auto_now_add=True)
      updated = models.DateTimeField(auto_now=True)
  
      class Meta:
          abstract = True
  
      def __str__(self):
          return self.name or self.client_id
  
      @property
      def default_redirect_uri(self):
          """
          Returns the default redirect_uri extracting the first item from
          the :attr:`redirect_uris` string
          """
          if self.redirect_uris:
              return self.redirect_uris.split().pop(0)
  
          assert False, (
              "If you are using implicit, authorization_code"
              "or all-in-one grant_type, you must define "
              "redirect_uris field in your Application model"
          )
  
      def redirect_uri_allowed(self, uri):
          """
          Checks if given url is one of the items in :attr:`redirect_uris` string
  
          :param uri: Url to check
          """
          parsed_uri = urlparse(uri)
          uqs_set = set(parse_qsl(parsed_uri.query))
          for allowed_uri in self.redirect_uris.split():
              parsed_allowed_uri = urlparse(allowed_uri)
  
              if (parsed_allowed_uri.scheme == parsed_uri.scheme and
                      parsed_allowed_uri.netloc == parsed_uri.netloc and
                      parsed_allowed_uri.path == parsed_uri.path):
  
                  aqs_set = set(parse_qsl(parsed_allowed_uri.query))
  
                  if aqs_set.issubset(uqs_set):
                      return True
  
          return False
  
      def clean(self):
          from django.core.exceptions import ValidationError
  
          grant_types = (
              AbstractApplication.GRANT_AUTHORIZATION_CODE,
              AbstractApplication.GRANT_IMPLICIT,
          )
  
          redirect_uris = self.redirect_uris.strip().split()
          allowed_schemes = set(s.lower() for s in self.get_allowed_schemes())
  
          if redirect_uris:
              validator = RedirectURIValidator(WildcardSet())
              for uri in redirect_uris:
                  validator(uri)
                  scheme = urlparse(uri).scheme
                  if scheme not in allowed_schemes:
                      raise ValidationError(_(
                          "Unauthorized redirect scheme: {scheme}"
                      ).format(scheme=scheme))
  
          elif self.authorization_grant_type in grant_types:
              raise ValidationError(_(
                  "redirect_uris cannot be empty with grant_type {grant_type}"
              ).format(grant_type=self.authorization_grant_type))
  
      def get_absolute_url(self):
          return reverse("oauth2_provider:detail", args=[str(self.id)])
  
      def get_allowed_schemes(self):
          """
          Returns the list of redirect schemes allowed by the Application.
          By default, returns `ALLOWED_REDIRECT_URI_SCHEMES`.
          """
          return oauth2_settings.ALLOWED_REDIRECT_URI_SCHEMES
  
      def allows_grant_type(self, *grant_types):
          return self.authorization_grant_type in grant_types
  
      def is_usable(self, request):
          """
          Determines whether the application can be used.
  
          :param request: The oauthlib.common.Request being processed.
          """
          return True
  
  
  class ApplicationManager(models.Manager):
      def get_by_natural_key(self, client_id):
          return self.get(client_id=client_id)
  
  
  class Application(AbstractApplication):
      objects = ApplicationManager()
  
      class Meta(AbstractApplication.Meta):
          swappable = "OAUTH2_PROVIDER_APPLICATION_MODEL"
  
      def natural_key(self):
          return (self.client_id,)
  
  
  class AbstractGrant(models.Model):
      """
      A Grant instance represents a token with a short lifetime that can
      be swapped for an access token, as described in :rfc:`4.1.2`
  
      Fields:
  
      * :attr:`user` The Django user who requested the grant
      * :attr:`code` The authorization code generated by the authorization server
      * :attr:`application` Application instance this grant was asked for
      * :attr:`expires` Expire time in seconds, defaults to
                        :data:`settings.AUTHORIZATION_CODE_EXPIRE_SECONDS`
      * :attr:`redirect_uri` Self explained
      * :attr:`scope` Required scopes, optional
      * :attr:`code_challenge` PKCE code challenge
      * :attr:`code_challenge_method` PKCE code challenge transform algorithm
      """
      CODE_CHALLENGE_PLAIN = "plain"
      CODE_CHALLENGE_S256 = "S256"
      CODE_CHALLENGE_METHODS = (
          (CODE_CHALLENGE_PLAIN, "plain"),
          (CODE_CHALLENGE_S256, "S256")
      )
  
      id = models.BigAutoField(primary_key=True)
      user = models.ForeignKey(
          settings.AUTH_USER_MODEL, on_delete=models.CASCADE,
          related_name="%(app_label)s_%(class)s"
      )
      code = models.CharField(max_length=255, unique=True)  # code comes from oauthlib
      application = models.ForeignKey(
          oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE
      )
      expires = models.DateTimeField()
      redirect_uri = models.CharField(max_length=255)
      scope = models.TextField(blank=True)
  
      created = models.DateTimeField(auto_now_add=True)
      updated = models.DateTimeField(auto_now=True)
  
      code_challenge = models.CharField(max_length=128, blank=True, default="")
      code_challenge_method = models.CharField(
          max_length=10, blank=True, default="", choices=CODE_CHALLENGE_METHODS)
  
      def is_expired(self):
          """
          Check token expiration with timezone awareness
          """
          if not self.expires:
              return True
  
          return timezone.now() >= self.expires
  
      def redirect_uri_allowed(self, uri):
          return uri == self.redirect_uri
  
      def __str__(self):
          return self.code
  
      class Meta:
          abstract = True
  
  
  class Grant(AbstractGrant):
      class Meta(AbstractGrant.Meta):
          swappable = "OAUTH2_PROVIDER_GRANT_MODEL"
  
  
  class AbstractAccessToken(models.Model):
      """
      An AccessToken instance represents the actual access token to
      access user's resources, as in :rfc:`5`.
  
      Fields:
  
      * :attr:`user` The Django user representing resources" owner
      * :attr:`source_refresh_token` If from a refresh, the consumed RefeshToken
      * :attr:`token` Access token
      * :attr:`application` Application instance
      * :attr:`expires` Date and time of token expiration, in DateTime format
      * :attr:`scope` Allowed scopes
      """
      id = models.BigAutoField(primary_key=True)
      user = models.ForeignKey(
          settings.AUTH_USER_MODEL, on_delete=models.CASCADE, blank=True, null=True,
          related_name="%(app_label)s_%(class)s"
      )
      source_refresh_token = models.OneToOneField(
          # unique=True implied by the OneToOneField
          oauth2_settings.REFRESH_TOKEN_MODEL, on_delete=models.SET_NULL, blank=True, null=True,
          related_name="refreshed_access_token"
      )
      token = models.CharField(max_length=255, unique=True, )
      application = models.ForeignKey(
          oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE, blank=True, null=True,
      )
      expires = models.DateTimeField()
      scope = models.TextField(blank=True)
  
      created = models.DateTimeField(auto_now_add=True)
      updated = models.DateTimeField(auto_now=True)
  
      def is_valid(self, scopes=None):
          """
          Checks if the access token is valid.
  
          :param scopes: An iterable containing the scopes to check or None
          """
          return not self.is_expired() and self.allow_scopes(scopes)
  
      def is_expired(self):
          """
          Check token expiration with timezone awareness
          """
          if not self.expires:
              return True
  
          return timezone.now() >= self.expires
  
      def allow_scopes(self, scopes):
          """
          Check if the token allows the provided scopes
  
          :param scopes: An iterable containing the scopes to check
          """
          if not scopes:
              return True
  
          provided_scopes = set(self.scope.split())
          resource_scopes = set(scopes)
  
          return resource_scopes.issubset(provided_scopes)
  
      def revoke(self):
          """
          Convenience method to uniform tokens" interface, for now
          simply remove this token from the database in order to revoke it.
          """
          self.delete()
  
      @property
      def scopes(self):
          """
          Returns a dictionary of allowed scope names (as keys) with their descriptions (as values)
          """
          all_scopes = get_scopes_backend().get_all_scopes()
          token_scopes = self.scope.split()
          return {name: desc for name, desc in all_scopes.items() if name in token_scopes}
  
      def __str__(self):
          return self.token
  
      class Meta:
          abstract = True
  
  
  class AccessToken(AbstractAccessToken):
      class Meta(AbstractAccessToken.Meta):
          swappable = "OAUTH2_PROVIDER_ACCESS_TOKEN_MODEL"
  
  
  class AbstractRefreshToken(models.Model):
      """
      A RefreshToken instance represents a token that can be swapped for a new
      access token when it expires.
  
      Fields:
  
      * :attr:`user` The Django user representing resources" owner
      * :attr:`token` Token value
      * :attr:`application` Application instance
      * :attr:`access_token` AccessToken instance this refresh token is
                             bounded to
      * :attr:`revoked` Timestamp of when this refresh token was revoked
      """
      id = models.BigAutoField(primary_key=True)
      user = models.ForeignKey(
          settings.AUTH_USER_MODEL, on_delete=models.CASCADE,
          related_name="%(app_label)s_%(class)s"
      )
      token = models.CharField(max_length=255)
      application = models.ForeignKey(
          oauth2_settings.APPLICATION_MODEL, on_delete=models.CASCADE)
      access_token = models.OneToOneField(
          oauth2_settings.ACCESS_TOKEN_MODEL, on_delete=models.SET_NULL, blank=True, null=True,
          related_name="refresh_token"
      )
  
      created = models.DateTimeField(auto_now_add=True)
      updated = models.DateTimeField(auto_now=True)
      revoked = models.DateTimeField(null=True)
  
      def revoke(self):
          """
          Mark this refresh token revoked and revoke related access token
          """
          access_token_model = get_access_token_model()
          refresh_token_model = get_refresh_token_model()
          with transaction.atomic():
              self = refresh_token_model.objects.filter(
                  pk=self.pk, revoked__isnull=True
              ).select_for_update().first()
              if not self:
                  return
  
              try:
                  access_token_model.objects.get(id=self.access_token_id).revoke()
              except access_token_model.DoesNotExist:
                  pass
              self.access_token = None
              self.revoked = timezone.now()
              self.save()
  
      def __str__(self):
          return self.token
  
      class Meta:
          abstract = True
          unique_together = ("token", "revoked",)
  
  
  class RefreshToken(AbstractRefreshToken):
      class Meta(AbstractRefreshToken.Meta):
          swappable = "OAUTH2_PROVIDER_REFRESH_TOKEN_MODEL"
  
  
  def get_application_model():
      """ Return the Application model that is active in this project. """
      return apps.get_model(oauth2_settings.APPLICATION_MODEL)
  
  
  def get_grant_model():
      """ Return the Grant model that is active in this project. """
      return apps.get_model(oauth2_settings.GRANT_MODEL)
  
  
  def get_access_token_model():
      """ Return the AccessToken model that is active in this project. """
      return apps.get_model(oauth2_settings.ACCESS_TOKEN_MODEL)
  
  
  def get_refresh_token_model():
      """ Return the RefreshToken model that is active in this project. """
      return apps.get_model(oauth2_settings.REFRESH_TOKEN_MODEL)
  
  
  def clear_expired():
      now = timezone.now()
      refresh_expire_at = None
      access_token_model = get_access_token_model()
      refresh_token_model = get_refresh_token_model()
      grant_model = get_grant_model()
      REFRESH_TOKEN_EXPIRE_SECONDS = oauth2_settings.REFRESH_TOKEN_EXPIRE_SECONDS
      if REFRESH_TOKEN_EXPIRE_SECONDS:
          if not isinstance(REFRESH_TOKEN_EXPIRE_SECONDS, timedelta):
              try:
                  REFRESH_TOKEN_EXPIRE_SECONDS = timedelta(seconds=REFRESH_TOKEN_EXPIRE_SECONDS)
              except TypeError:
                  e = "REFRESH_TOKEN_EXPIRE_SECONDS must be either a timedelta or seconds"
                  raise ImproperlyConfigured(e)
          refresh_expire_at = now - REFRESH_TOKEN_EXPIRE_SECONDS
  
      with transaction.atomic():
          if refresh_expire_at:
              revoked = refresh_token_model.objects.filter(
                  revoked__lt=refresh_expire_at,
              )
              expired = refresh_token_model.objects.filter(
                  access_token__expires__lt=refresh_expire_at,
              )
  
              logger.info("%s Revoked refresh tokens to be deleted", revoked.count())
              logger.info("%s Expired refresh tokens to be deleted", expired.count())
  
              revoked.delete()
              expired.delete()
          else:
              logger.info("refresh_expire_at is %s. No refresh tokens deleted.",
                          refresh_expire_at)
  
          access_tokens = access_token_model.objects.filter(
              refresh_token__isnull=True,
              expires__lt=now
          )
          grants = grant_model.objects.filter(expires__lt=now)
  
          logger.info("%s Expired access tokens to be deleted", access_tokens.count())
          logger.info("%s Expired grant tokens to be deleted", grants.count())
  
          access_tokens.delete()
          grants.delete()
  