make userdbs admin work properly
* add userdbs app docstring * add userdbs.app.UserdbsAppConfig * implement userdbs.admin.DatabaseUserCreationForm, userdbs.admin.UserDatabaseCreationForm, userdbs.admin.DatabaseUserAdmin, userdbs.admin.UserDatabaseAdmin * add docstrings to userdbs.models * rename userdbs.models.DatabaseUserManager._get_next_username to _get_next_dbuser_name * fix format string issues in userdbs.models.DatabaseUserManager and userdbs.UserDatabaseManager._get_next_dbname * delete related databases in userdbs.models.UserDatabase.delete
This commit is contained in:
parent
be4ea9cc77
commit
6edbe17a3b
|
|
@ -0,0 +1,5 @@
|
|||
"""
|
||||
This app is for managing database users and user databases.
|
||||
|
||||
"""
|
||||
default_app_config = 'userdbs.apps.UserdbsAppConfig'
|
||||
|
|
@ -4,7 +4,9 @@ Admin functionality for the :py:mod:`userdbs.models` models.
|
|||
"""
|
||||
from __future__ import absolute_import
|
||||
|
||||
from django import forms
|
||||
from django.contrib import admin
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
from .models import (
|
||||
DatabaseUser,
|
||||
|
|
@ -12,5 +14,266 @@ from .models import (
|
|||
)
|
||||
|
||||
|
||||
admin.site.register(DatabaseUser)
|
||||
admin.site.register(UserDatabase)
|
||||
class DatabaseUserCreationForm(forms.ModelForm):
|
||||
"""
|
||||
A form for creating :py:class:`database users
|
||||
<userdbs.models.DatabaseUser>`
|
||||
|
||||
"""
|
||||
|
||||
class Meta:
|
||||
model = DatabaseUser
|
||||
fields = ['osuser', 'db_type']
|
||||
|
||||
def save(self, commit=True):
|
||||
"""
|
||||
Save the database user.
|
||||
|
||||
:param boolean commit: whether to save the created database user
|
||||
:return: database user instance
|
||||
:rtype: :py:class:`userdbs.models.DatabaseUser`
|
||||
|
||||
"""
|
||||
dbuser = DatabaseUser.objects.create_database_user(
|
||||
osuser=self.cleaned_data['osuser'],
|
||||
db_type=self.cleaned_data['db_type'], commit=commit)
|
||||
return dbuser
|
||||
|
||||
def save_m2m(self):
|
||||
"""
|
||||
Noop.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
class UserDatabaseCreationForm(forms.ModelForm):
|
||||
"""
|
||||
A form for creating :py:class:`user databases
|
||||
<userdbs.models.UserDatabase>`
|
||||
|
||||
"""
|
||||
|
||||
class Meta:
|
||||
model = UserDatabase
|
||||
fields = ['db_user']
|
||||
|
||||
def save(self, commit=True):
|
||||
"""
|
||||
Save the user database.
|
||||
|
||||
:param boolean commit: whether to save the created user database
|
||||
:return: user database instance
|
||||
:rtype: :py:class:`userdbs.models.UserDatabase`
|
||||
|
||||
"""
|
||||
database = UserDatabase.objects.create_userdatabase(
|
||||
db_user=self.cleaned_data['db_user'], commit=commit)
|
||||
return database
|
||||
|
||||
def save_m2m(self):
|
||||
"""
|
||||
Noop.
|
||||
|
||||
"""
|
||||
|
||||
|
||||
class DatabaseUserAdmin(admin.ModelAdmin):
|
||||
"""
|
||||
Admin class for working with :py:class:`database users
|
||||
<userdbs.models.DatabaseUser>`
|
||||
|
||||
"""
|
||||
actions = ['perform_delete_selected']
|
||||
add_form = DatabaseUserCreationForm
|
||||
|
||||
def get_form(self, request, obj=None, **kwargs):
|
||||
"""
|
||||
Use special form for database user creation.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: either a :py:class:`Database user
|
||||
<userdbs.models.DatabaseUser>` instance or None for a new database
|
||||
user
|
||||
:param kwargs: keyword arguments to be passed to
|
||||
:py:meth:`django.contrib.admin.ModelAdmin.get_form`
|
||||
:return: form instance
|
||||
|
||||
"""
|
||||
defaults = {}
|
||||
if obj is None:
|
||||
defaults.update({
|
||||
'form': self.add_form,
|
||||
})
|
||||
defaults.update(kwargs)
|
||||
return super(DatabaseUserAdmin, self).get_form(
|
||||
request, obj, **defaults)
|
||||
|
||||
def get_readonly_fields(self, request, obj=None):
|
||||
"""
|
||||
Make sure that osuser, name and db_type are not editable for existing
|
||||
database users.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: either a :py:class:`Database user
|
||||
<userdbs.models.DatabaseUser>` instance or None for a new database
|
||||
user
|
||||
:return: a list of fields
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
if obj:
|
||||
return ['osuser', 'name', 'db_type']
|
||||
return []
|
||||
|
||||
def save_model(self, request, obj, form, change):
|
||||
"""
|
||||
Make sure that the user is created in the target database.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: a :py:class:`Database user <userdbs.models.DatabaseUser>`
|
||||
instance
|
||||
:param form: the form instance
|
||||
:param boolean change: whether this is a change operation or not
|
||||
|
||||
"""
|
||||
if not change:
|
||||
obj.create_in_database()
|
||||
super(DatabaseUserAdmin, self).save_model(request, obj, form, change)
|
||||
|
||||
def perform_delete_selected(self, request, queryset):
|
||||
"""
|
||||
Action to delete a list of selected database users.
|
||||
|
||||
This action calls the delete method of each selected database user in
|
||||
contrast to the default `delete_selected`
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param queryset: Django ORM queryset representing the selected database
|
||||
users
|
||||
|
||||
"""
|
||||
for dbuser in queryset.all():
|
||||
dbuser.delete()
|
||||
perform_delete_selected.short_description = _(
|
||||
'Delete selected database users')
|
||||
|
||||
def get_actions(self, request):
|
||||
"""
|
||||
Get the available actions for database users.
|
||||
|
||||
This overrides the default behavior to remove the default
|
||||
`delete_selected` action.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:return: list of actions
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
actions = super(DatabaseUserAdmin, self).get_actions(request)
|
||||
if 'delete_selected' in actions:
|
||||
del actions['delete_selected']
|
||||
return actions
|
||||
|
||||
|
||||
class UserDatabaseAdmin(admin.ModelAdmin):
|
||||
"""
|
||||
Admin class for working with :py:class:`user databases
|
||||
<userdbs.models.UserDatabase>`
|
||||
|
||||
"""
|
||||
actions = ['perform_delete_selected']
|
||||
add_form = UserDatabaseCreationForm
|
||||
|
||||
def get_form(self, request, obj=None, **kwargs):
|
||||
"""
|
||||
Use special form for user database creation.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: either a :py:class:`User database
|
||||
<userdbs.models.UserDatabase>` instance or None for a new user
|
||||
database
|
||||
:param kwargs: keyword arguments to be passed to
|
||||
:py:meth:`django.contrib.admin.ModelAdmin.get_form`
|
||||
:return: form instance
|
||||
|
||||
"""
|
||||
defaults = {}
|
||||
if obj is None:
|
||||
defaults.update({
|
||||
'form': self.add_form,
|
||||
})
|
||||
defaults.update(kwargs)
|
||||
return super(UserDatabaseAdmin, self).get_form(
|
||||
request, obj, **defaults)
|
||||
|
||||
def get_readonly_fields(self, request, obj=None):
|
||||
"""
|
||||
Make sure that db_name and db_user are not editable for existing user
|
||||
databases.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: either a :py:class:`User database
|
||||
<userdbs.models.UserDatabase>` instance or None for a new user
|
||||
database
|
||||
:return: a list of fields
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
if obj:
|
||||
return ['db_name', 'db_user']
|
||||
return []
|
||||
|
||||
def save_model(self, request, obj, form, change):
|
||||
"""
|
||||
Make sure that the database is created in the target database server.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param obj: a :py:class:`Database user <userdbs.models.DatabaseUser>`
|
||||
instance
|
||||
:param form: the form instance
|
||||
:param boolean change: whether this is a change operation or not
|
||||
|
||||
"""
|
||||
if not change:
|
||||
obj.create_in_database()
|
||||
super(UserDatabaseAdmin, self).save_model(request, obj, form, change)
|
||||
|
||||
def perform_delete_selected(self, request, queryset):
|
||||
"""
|
||||
Action to delete a list of selected user databases.
|
||||
|
||||
This action calls the delete method of each selected user database in
|
||||
contrast to the default `delete_selected`
|
||||
|
||||
:param request: the current HTTP request
|
||||
:param queryset: Django ORM queryset representing the selected user
|
||||
databases
|
||||
|
||||
"""
|
||||
for dbuser in queryset.all():
|
||||
dbuser.delete()
|
||||
for database in queryset.all():
|
||||
database.delete()
|
||||
perform_delete_selected.short_description = _(
|
||||
'Delete selected user databases')
|
||||
|
||||
def get_actions(self, request):
|
||||
"""
|
||||
Get the available actions for user databases.
|
||||
|
||||
This overrides the default behavior to remove the default
|
||||
`delete_selected` action.
|
||||
|
||||
:param request: the current HTTP request
|
||||
:return: list of actions
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
actions = super(UserDatabaseAdmin, self).get_actions(request)
|
||||
if 'delete_selected' in actions:
|
||||
del actions['delete_selected']
|
||||
return actions
|
||||
|
||||
|
||||
admin.site.register(DatabaseUser, DatabaseUserAdmin)
|
||||
admin.site.register(UserDatabase, UserDatabaseAdmin)
|
||||
|
|
|
|||
18
gnuviechadmin/userdbs/apps.py
Normal file
18
gnuviechadmin/userdbs/apps.py
Normal file
|
|
@ -0,0 +1,18 @@
|
|||
"""
|
||||
This module contains the :py:class:`django.apps.AppConfig` instance for the
|
||||
:py:mod:`userdbs` app.
|
||||
|
||||
"""
|
||||
from __future__ import unicode_literals
|
||||
|
||||
from django.apps import AppConfig
|
||||
from django.utils.translation import ugettext_lazy as _
|
||||
|
||||
|
||||
class UserdbsAppConfig(AppConfig):
|
||||
"""
|
||||
AppConfig for the :py:mod:`userdbs` app.
|
||||
|
||||
"""
|
||||
name = 'userdbs'
|
||||
verbose_name = _('Database Users and their Databases')
|
||||
|
|
@ -43,7 +43,7 @@ class DatabaseUserManager(models.Manager):
|
|||
|
||||
"""
|
||||
|
||||
def _get_next_username(self, osuser, db_type):
|
||||
def _get_next_dbuser_name(self, osuser, db_type):
|
||||
"""
|
||||
Get the next available database user name.
|
||||
|
||||
|
|
@ -53,15 +53,15 @@ class DatabaseUserManager(models.Manager):
|
|||
:rtype: str
|
||||
"""
|
||||
count = 1
|
||||
db_username_format = "{0}db{{1:02d}}".format(osuser.username)
|
||||
nextname = db_username_format.format(count)
|
||||
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 = db_username_format.format(count)
|
||||
nextname = dbuser_name_format.format(count)
|
||||
else:
|
||||
break
|
||||
return nextname
|
||||
|
|
@ -78,7 +78,7 @@ class DatabaseUserManager(models.Manager):
|
|||
: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: password for the user
|
||||
:param str password: initial password or None
|
||||
:param boolean commit: whether the user should be persisted
|
||||
:return: :py:class:`userdbs.models.DatabaseUser` instance
|
||||
|
||||
|
|
@ -89,14 +89,11 @@ class DatabaseUserManager(models.Manager):
|
|||
|
||||
"""
|
||||
if username is None:
|
||||
username = self._get_next_username(osuser, db_type)
|
||||
if password is None:
|
||||
password = generate_password()
|
||||
username = self._get_next_dbuser_name(osuser, db_type)
|
||||
db_user = DatabaseUser(
|
||||
osuser=osuser, db_type=db_type, username=username)
|
||||
osuser=osuser, db_type=db_type, name=username, password=password)
|
||||
if commit:
|
||||
db_user.create_in_database(password)
|
||||
# TODO: send GPG encrypted mail with this information
|
||||
db_user.create_in_database()
|
||||
db_user.save()
|
||||
return db_user
|
||||
|
||||
|
|
@ -120,16 +117,19 @@ class DatabaseUser(TimeStampedModel, models.Model):
|
|||
def __str__(self):
|
||||
return "%(name)s (%(db_type)s for %(osuser)s)" % {
|
||||
'name': self.name,
|
||||
'db_type': self.db_type,
|
||||
'db_type': self.get_db_type_display(),
|
||||
'osuser': self.osuser.username,
|
||||
}
|
||||
|
||||
def create_in_database(self, password):
|
||||
def create_in_database(self, password=None):
|
||||
"""
|
||||
Create this user in the target database.
|
||||
|
||||
:param str password: initial password for the database user
|
||||
"""
|
||||
if password is None:
|
||||
password = generate_password()
|
||||
# TODO: send GPG encrypted mail with this information
|
||||
if self.db_type == DB_TYPES.pgsql:
|
||||
create_pgsql_user.delay(self.name, password).get()
|
||||
elif self.db_type == DB_TYPES.mysql:
|
||||
|
|
@ -144,9 +144,9 @@ class DatabaseUser(TimeStampedModel, models.Model):
|
|||
:param str password: new password for the database user
|
||||
"""
|
||||
if self.db_type == DB_TYPES.pgsql:
|
||||
set_pgsql_userpassword.delay(self.name, password).get()
|
||||
set_pgsql_userpassword.delay(self.name, password).get(timeout=5)
|
||||
elif self.db_type == DB_TYPES.mysql:
|
||||
set_mysql_userpassword.delay(self.name, password).get()
|
||||
set_mysql_userpassword.delay(self.name, password).get(timeout=5)
|
||||
else:
|
||||
raise ValueError('Unknown database type %d' % self.db_type)
|
||||
|
||||
|
|
@ -162,10 +162,12 @@ class DatabaseUser(TimeStampedModel, models.Model):
|
|||
:py:meth:`django.db.models.Model.delete`
|
||||
|
||||
"""
|
||||
for database in self.userdatabase_set.all():
|
||||
database.delete()
|
||||
if self.db_type == DB_TYPES.pgsql:
|
||||
delete_pgsql_user.delay(self.name).get()
|
||||
delete_pgsql_user.delay(self.name).get(propagate=False, timeout=5)
|
||||
elif self.db_type == DB_TYPES.mysql:
|
||||
delete_mysql_user.delay(self.name).get()
|
||||
delete_mysql_user.delay(self.name).get(propagate=False, timeout=5)
|
||||
else:
|
||||
raise ValueError('Unknown database type %d' % self.db_type)
|
||||
super(DatabaseUser, self).delete(*args, **kwargs)
|
||||
|
|
@ -187,7 +189,7 @@ class UserDatabaseManager(models.Manager):
|
|||
|
||||
"""
|
||||
count = 1
|
||||
db_name_format = "{0}_{{1:02d}}".format(db_user.name)
|
||||
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(
|
||||
|
|
@ -216,7 +218,6 @@ class UserDatabaseManager(models.Manager):
|
|||
database = UserDatabase(db_user=db_user, db_name=db_name)
|
||||
if commit:
|
||||
database.create_in_database()
|
||||
# TODO: send GPG encrypted mail with this information
|
||||
database.save()
|
||||
return database
|
||||
|
||||
|
|
@ -228,6 +229,8 @@ class UserDatabase(TimeStampedModel, models.Model):
|
|||
_('database name'), max_length=63)
|
||||
db_user = models.ForeignKey(DatabaseUser, verbose_name=_('database user'))
|
||||
|
||||
objects = UserDatabaseManager()
|
||||
|
||||
class Meta:
|
||||
unique_together = ['db_name', 'db_user']
|
||||
verbose_name = _('user database')
|
||||
|
|
@ -244,6 +247,7 @@ class UserDatabase(TimeStampedModel, models.Model):
|
|||
Create this database (schema) in the target database.
|
||||
|
||||
"""
|
||||
# TODO: send GPG encrypted mail with this information
|
||||
if self.db_user.db_type == DB_TYPES.pgsql:
|
||||
create_pgsql_database.delay(self.db_name, self.db_user.name).get()
|
||||
elif self.db_user.db_type == DB_TYPES.mysql:
|
||||
|
|
@ -263,9 +267,9 @@ class UserDatabase(TimeStampedModel, models.Model):
|
|||
:py:meth:`django.db.models.Model.delete`
|
||||
|
||||
"""
|
||||
if self.db_type == DB_TYPES.pgsql:
|
||||
if self.db_user.db_type == DB_TYPES.pgsql:
|
||||
delete_pgsql_database.delay(self.db_name, self.db_user.name).get()
|
||||
elif self.db_type == DB_TYPES.mysql:
|
||||
elif self.db_user.db_type == DB_TYPES.mysql:
|
||||
delete_mysql_database.delay(self.db_name, self.db_user.name).get()
|
||||
else:
|
||||
raise ValueError('Unknown database type %d' % self.db_type)
|
||||
|
|
|
|||
Loading…
Reference in a new issue