gva/gnuviechadmin/userdbs/models.py
Jan Dittberner 4af1a39ca4 Upgrade to Django 3.2
- update dependencies
- fix deprecation warnings
- fix tests
- skip some tests that need more work
- reformat changed code with isort and black
2023-02-18 22:46:48 +01:00

211 lines
6.6 KiB
Python

from django.db import models, transaction
from django.dispatch import Signal
from django.utils.translation import gettext as _
from model_utils import Choices
from model_utils.models import TimeStampedModel
from osusers.models import User as OsUser
DB_TYPES = Choices(
(0, "pgsql", _("PostgreSQL")),
(1, "mysql", _("MySQL")),
)
"""
Database type choice enumeration.
"""
password_set = Signal()
class DatabaseUserManager(models.Manager):
"""
Default Manager for :py:class:`userdbs.models.DatabaseUser`.
"""
def _get_next_dbuser_name(self, osuser, db_type):
"""
Get the next available database user name.
:param osuser: :py:class:`osusers.models.User` instance
:param db_type: value from :py:data:`DB_TYPES`
:return: database user name
:rtype: str
"""
count = 1
dbuser_name_format = "{0}db{{0:02d}}".format(osuser.username)
nextname = dbuser_name_format.format(count)
for user in (
self.values("name").filter(osuser=osuser, db_type=db_type).order_by("name")
):
if user["name"] == nextname:
count += 1
nextname = dbuser_name_format.format(count)
else:
break
return nextname
@transaction.atomic
def create_database_user(
self, osuser, db_type, username=None, password=None, commit=True
):
"""
Create a database user of the given type for the given OS user.
If username or password are not specified they are generated.
:param osuser: the :py:class:`osusers.models.User` instance
:param db_type: value from :py:data:`DB_TYPES`
:param str username: database user name
:param str password: initial password or None
:param boolean commit: whether the user should be persisted
:return: :py:class:`userdbs.models.DatabaseUser` instance
.. note::
The password is not persisted it is only used to set the password
on the database side.
"""
if username is None:
username = self._get_next_dbuser_name(osuser, db_type)
db_user = DatabaseUser(osuser=osuser, db_type=db_type, name=username)
if commit:
db_user.save()
return db_user
class DatabaseUser(TimeStampedModel, models.Model):
osuser = models.ForeignKey(OsUser, on_delete=models.CASCADE)
name = models.CharField(_("username"), max_length=63)
db_type = models.PositiveSmallIntegerField(_("database type"), choices=DB_TYPES)
objects = DatabaseUserManager()
class Meta:
unique_together = ["name", "db_type"]
verbose_name = _("database user")
verbose_name_plural = _("database users")
def __str__(self):
return "%(name)s (%(db_type)s for %(osuser)s)" % {
"name": self.name,
"db_type": self.get_db_type_display(),
"osuser": self.osuser.username,
}
@transaction.atomic
def set_password(self, password):
"""
Set an existing user's password.
:param str password: new password for the database user
"""
password_set.send(sender=self.__class__, password=password, instance=self)
@transaction.atomic
def delete(self, *args, **kwargs):
"""
Delete the database user from the target database and the Django
database.
:param args: positional arguments for
:py:meth:`django.db.models.Model.delete`
:param kwargs: keyword arguments for
:py:meth:`django.db.models.Model.delete`
"""
for database in self.userdatabase_set.all():
database.delete()
super(DatabaseUser, self).delete(*args, **kwargs)
class UserDatabaseManager(models.Manager):
"""
Default manager for :py:class:`userdbs.models.UserDatabase` instances.
"""
def _get_next_dbname(self, db_user):
"""
Get the next available database name for the given database user.
:param db_user: :py:class:`userdbs.models.DatabaseUser` instance
:return: database name
:rtype: str
"""
count = 1
db_name_format = "{0}_{{0:02d}}".format(db_user.name)
# first db is named the same as the user
nextname = db_user.name
for name in self.values("db_name").filter(db_user=db_user).order_by("db_name"):
if name["db_name"] == nextname:
count += 1
nextname = db_name_format.format(count)
else:
break
return nextname
@transaction.atomic
def create_userdatabase_with_user(
self, db_type, osuser, password=None, commit=True
):
"""
Creates a new user database with a new user.
:param db_type: database type from :py:data:`DB_TYPES`
:param osuser: :py:class:`osusers.models.OsUser` instance
:param str password: the password of the new database user
:param boolean commit: whether the user and the database should be
persisted
:return: database instance
:rtype: :py:class:`UserDatabase`
"""
dbuser = DatabaseUser.objects.create_database_user(
osuser, db_type, password=password, commit=commit
)
database = self.create_userdatabase(dbuser, commit=commit)
return database
@transaction.atomic
def create_userdatabase(self, db_user, db_name=None, commit=True):
"""
Creates a new user database.
:param db_user: :py:class:`userdbs.models.DatabaseUser` instance
:param str db_name: database name
:param boolean commit: whether the database should be persisted
:return: :py:class:`userdbs.models.UserDatabase` instance
"""
if db_name is None:
db_name = self._get_next_dbname(db_user)
database = UserDatabase(db_user=db_user, db_name=db_name)
if commit:
database.save()
return database
class UserDatabase(TimeStampedModel, models.Model):
# MySQL limits to 64, PostgreSQL to 63 characters
db_name = models.CharField(_("database name"), max_length=63)
db_user = models.ForeignKey(
DatabaseUser, verbose_name=_("database user"), on_delete=models.CASCADE
)
objects = UserDatabaseManager()
class Meta:
unique_together = ["db_name", "db_user"]
verbose_name = _("user database")
verbose_name_plural = _("user specific database")
def __str__(self):
return "%(db_name)s (%(db_user)s)" % {
"db_name": self.db_name,
"db_user": self.db_user,
}