Produits Entr'ouvert » Django Journal

from . import models

def export_as_csv_generator(queryset):

    header = ['time', 'tag', 'message']

    tags = set(models.Tag.objects.filter(objectdata__journal__in=queryset).values_list('name', flat=True))

        tags.add('%s__id' % tag)

    tags |= set(models.Tag.objects.filter(stringdata__journal__in=queryset).values_list('name', flat=True))

    extra_headers = list(sorted(tags))

    for journal in queryset:

        row = {

        for stringdata in journal.stringdata_set.all():

            row_name = stringdata.tag.name.encode('utf-8')

            row[force_text(row_name)] = force_text(stringdata.content)

                row[row_name] = force_text(objectdata.content_object)

        yield row

def export_as_csv(modeladmin, request, queryset):

    """

    CSV export for journal

    for row in l:

        writer.writerow(row)

    return response

export_as_csv.short_description = _(u"Export CSV file")

class ModelAdminFormatter(Formatter):

        self.filter_link = filter_link

        self.object_link = object_link

        self.model_admin = model_admin

    def build_object_link(self, value):

        content_type = ContentType.objects.get_for_model(value.__class__)

        try:

            url = reverse(url, args=(value.pk,))

        except NoReverseMatch:

            if self.filter_link:

                content_type = ContentType.objects.get_for_model(value.__class__)

                res = u'<a href="?objectdata__content_type={0}&objectdata__object_id={1}">{2}</a>'.format(

            else:

                res = escape(force_text(value))

            if self.object_link:

    extra = 0

    max_num = 0

class StringDataInlineAdmin(admin.TabularInline):

    model = StringData

    fields = ('tag', 'content')

    extra = 0

    max_num = 0

class JournalAdmin(admin.ModelAdmin):

    list_display = ('time', '_tag', 'user', 'ip', 'message_for_list')

    list_filter = ('tag',)

    fields = ('time', 'tag', 'user', 'ip', 'message_for_change')

    readonly_fields = fields

    inlines = (

    )

    date_hierarchy = 'time'

    class Media:

        css = {

        }

    def queryset(self, request):

        '''Get as much data as possible using the fewest requests possible.'''

        qs = super(JournalAdmin, self).queryset(request)

        return qs

    def lookup_allowed(self, key, *args, **kwargs):

    def _tag(self, entry):

        name = entry.tag.name.replace(u'-', u'\u2011')

        return res

    _tag.short_description = _('tag')

    def ip(self, entry):

        '''Search and return any associated stringdata whose tag is "ip"'''

        for stringdata in entry.stringdata_set.all():

            if stringdata.tag.name == 'ip':

        return _('None')

    ip.short_description = _('IP')

    def user(self, entry):

        '''Search and return any associated objectdata whose tag is "user"'''

        for objectdata in entry.objectdata_set.all():

            if objectdata.tag.name == 'user':

        return _('None')

    user.short_description = _('User')

    def object_filter_link(self, objectdata):

            caption = force_text(objectdata.content_object)

        else:

            caption = _(u'<deleted {content_type} {object_id}>').format(

        return u'<a href="?objectdata__content_type={0}&objectdata__object_id={1}">{2}</a>'.format(

    def object_link(self, obj_data):

        if obj_data.content_object is None:

            return u''

        try:

            url = reverse(url, args=(obj_data.object_id,))

        except NoReverseMatch:

        formatter = ModelAdminFormatter(model_admin=self, filter_link=False)

        message = formatter.format(escape(entry.template.content), **ctx)

        return format_html('<span>{}</span>', mark_safe(message))

    message_for_change.short_description = _('Message')

    def message_for_list(self, entry):

        formatter = ModelAdminFormatter(model_admin=self)

        message = formatter.format(entry.template.content, **ctx)

        return format_html('<span>{}</span>', mark_safe(message))

    message_for_list.short_description = _('Message')

    message_for_list.admin_order_field = 'message'

admin.site.register(Journal, JournalAdmin)

admin.site.register(Tag)

if hasattr(transaction, 'atomic'):

    atomic = transaction.atomic

else:

    class Transaction(object):

        sid = None

            def wrapper(*args, **kwargs):

                with self.__class__(using=self.using):

                    return func(*args, **kwargs)

    def atomic(using=None):

        """

        if callable(using):

            return Transaction(DEFAULT_DB_ALIAS)(using)

        return Transaction(using)

def unicode_truncate(s, length, encoding='utf-8'):

    encoded = s.encode(encoding)[:length]

    return encoded.decode(encoding, 'ignore')

@atomic

def record(tag, template, using=None, **kwargs):

    template = force_text(template)

    tag = Tag.objects.using(using).get_cached(name=tag)

    template = Template.objects.using(using).get_cached(content=template)

    try:

        message = template.content.format(**kwargs)

    except (KeyError, IndexError) as e:

    try:

        logger = logging.getLogger('django.journal.%s' % tag)

        if tag.name == 'error' or tag.name.startswith('error-'):

        try:

            logging.getLogger('django.journal').exception('Unable to log msg')

        except:

    for name, value in kwargs.items():

        if value is None:

            continue

        tag = Tag.objects.using(using).get_cached(name=name)

        if isinstance(value, django.db.models.Model):

            journal.objectdata_set.create(

            )

        else:

            journal.stringdata_set.create(tag=tag, content=force_text(value))

def error_record(tag, tpl, **kwargs):

    if kwargs.get('using') is None:

        kwargs['using'] = getattr(settings, 'JOURNAL_DB_FOR_ERROR_ALIAS', 'default')

        '''Return Journal records linked to this object.'''

        content_type = ContentType.objects.get_for_model(obj)

        if tag is None:

        else:

            return self.filter(

    def for_objects(self, objects):

        if not objects:

            return self.none()

        if len(set(content_types)) != 1:

            raise ValueError('objects must have of the same content type')

    def for_tag(self, tag):

        from . import models

        if not isinstance(tag, models.Tag):

                return self.none()

        # always remember: multiple join (OR in WHERE) produces duplicate

        # lines ! Use .distinct() for safety.

class JournalManager(Manager.from_queryset(JournalQuerySet)):

    def get_query_set(self):

class JournalMiddleware(MiddlewareMixin):

    def process_request(self, request):

        user = getattr(request, 'user', None)

        ip = request.META.get('REMOTE_ADDR', None)

        def record(tag, template, using=None, **kwargs):

            if 'user' not in kwargs:

                kwargs['user'] = user

            if 'ip' not in kwargs:

                kwargs['ip'] = ip

        def error_record(tag, template, using=None, **kwargs):

            if 'user' not in kwargs:

                kwargs['user'] = user

            if 'ip' not in kwargs:

                kwargs['ip'] = ip

            journal.error_record(tag, template, using=using, **kwargs)

        request.record = record

        request.error_record = error_record

        return None

        migrations.CreateModel(

            name='Journal',

            fields=[

                ('time', models.DateTimeField(auto_now_add=True, verbose_name='time', db_index=True)),

                ('message', models.CharField(max_length=128, verbose_name='message', db_index=True)),

            ],

        migrations.CreateModel(

            name='ObjectData',

            fields=[

                ('object_id', models.PositiveIntegerField(verbose_name='object id', db_index=True)),

            ],

            options={

                'verbose_name': 'linked object',

        migrations.CreateModel(

            name='StringData',

            fields=[

                ('content', models.TextField(verbose_name='content')),

            ],

            options={

                'verbose_name': 'linked text string',

        migrations.CreateModel(

            name='Tag',

            fields=[

                ('name', models.CharField(unique=True, max_length=32, verbose_name='name', db_index=True)),

            ],

            options={

        migrations.CreateModel(

            name='Template',

            fields=[

                ('content', models.TextField(unique=True, verbose_name='content', db_index=True)),

            ],

            options={

        migrations.AddField(

            model_name='journal',

            name='tag',

        ),

        migrations.AddField(

            model_name='journal',

            name='template',

        ),

        migrations.AlterUniqueTogether(

            name='stringdata',

@python_2_unicode_compatible

class Tag(models.Model):

    objects = managers.TagManager()

    def __str__(self):

        return self.name

@python_2_unicode_compatible

class Template(models.Model):

    objects = managers.TemplateManager()

    def __str__(self):

        return self.content

@python_2_unicode_compatible

class Journal(models.Model):

    objects = managers.JournalManager()

    class Meta:

        ordering = ('-id',)

                ctx[data.tag.name] = data.content_object

            else:

                ctx[data.tag.name] = u'<deleted {content_type} {object_id}>'.format(

        for data in self.stringdata_set.all():

            ctx[data.tag.name] = data.content

        for text, field, format_spec, conversion in string.Formatter().parse(self.template.content):

        return ctx

    def add_object_tag(self, tag_name, obj):

    def __str__(self):

        ctx = self.message_context()

    def __repr__(self):

        return '<Journal pk:{0} tag:{1} message:{2}>'.format(

class StringData(models.Model):

    journal = models.ForeignKey(Journal, verbose_name=_('journal entry'), on_delete=models.CASCADE)

    tag = models.ForeignKey(Tag, verbose_name=_('tag'), on_delete=models.CASCADE)

    content = models.TextField(verbose_name=_('content'))

@python_2_unicode_compatible

class ObjectData(models.Model):

    journal = models.ForeignKey(Journal, verbose_name=_('journal entry'), on_delete=models.CASCADE)

    tag = models.ForeignKey(Tag, verbose_name=_('tag'), on_delete=models.CASCADE)

    class Meta:

        unique_together = (('journal', 'tag'),)

    def run(self):

        import os

        try:

            from django.core.management import call_command

        except ImportError:

        try:

            os.environ.pop('DJANGO_SETTINGS_MODULE', None)

            from django.core.management import call_command

            os.chdir(os.path.realpath('django_journal'))

            call_command('compilemessages')

        except ImportError:

def get_version():

    if os.path.exists('VERSION'):

        with open('VERSION', 'r') as v:

            return v.read()

    if os.path.exists('.git'):

        p = subprocess.Popen(

            ['git', 'describe', '--dirty=.dirty', '--match=v*'],

        result = p.communicate()[0]

        if p.returncode == 0:

            result = result.decode('ascii').strip()[1:]  # strip spaces/newlines and initial v

                version = result

            return version

        else:

    return '0.0'

INSTALLED_APPS = (

)

DATABASES = {

}

SECRET_KEY = "django_tests_secret_key"

        self.groups = []

        with transaction.atomic():

            for i in range(20):

            for i in range(20):

            for i in range(20):

                record('login', '{user} logged in', user=self.users[i])

            for i in range(20):

            for i in range(20):

                record('logout', '{user} logged out', user=self.users[i])

    def test_groups(self):

        for i, event in zip(range(40), Journal.objects.for_tag('group-changed').order_by('id')):

    def test_logout(self):

        for i, event in zip(range(20), Journal.objects.for_tag('logout').order_by('id')):

    def test_export_as_csv(self):

        qs = Journal.objects.all()

        l = list(export_as_csv_generator(qs))

        l = list(export_as_csv_generator(qs[:5]))

        self.assertEquals(set(l[0]), set(['time', 'tag', 'message', 'user', 'user__id']))

        for user in self.users: