MAINHACK

Viewing / Editing File: country.py

File is not writable. Editing disabled.
import time
from typing import Generator

import blinker
from peewee import (JOIN, CharField, Check, DoesNotExist, ForeignKeyField,
                    IntegerField)
from playhouse.shortcuts import model_to_dict

from defence360agent.model import instance, Model
from defence360agent.model.simplification import apply_order_by
from im360.internals import geo


class Country(Model):
    """Contains a single record per country, with its code and name."""
    id = CharField(primary_key=True, null=False)
    #: Country code, e.g. "US".
    code = CharField(max_length=2, unique=True, null=False)
    #: Country name, e.g. "United States".
    name = CharField(null=False)

    class Meta:
        db_table = 'country'
        database = instance.db

    @classmethod
    def update_from(cls, countries):
        """
        Load country info from iterable of dicts to database
        :param countries:
        :return:
        """
        for country in countries:
            cls.insert(country).on_conflict_replace().execute()


class CountryList(Model):
    """List of Countries in user WHITE/BLACK list."""

    #: Listname for blocked countries.
    BLACK = "BLACK"
    #: Listname for whitelisted countries (not officially supported).
    WHITE = "WHITE"

    #: The link to country code and name in :class:`Country`.
    country = ForeignKeyField(Country, primary_key=True, null=False)
    #: In which list all IPs from this country should be:
    #: :attr:`BLACK` or :attr:`WHITE`.
    listname = CharField(null=False,
                         constraints=[Check("listname in "
                                            "('WHITE','BLACK')")])
    #: Timestamp when the record was added.
    ctime = IntegerField(null=True, default=lambda: int(time.time()))
    #: Comment set by admin.
    comment = CharField(null=True)

    class Meta:
        db_table = 'country_list'
        database = instance.db

    class Signals:
        added = blinker.Signal()
        deleted = blinker.Signal()

    @classmethod
    def create(cls, **kwargs):
        obj = super().create(**kwargs)
        cls.Signals.added.send(obj.listname, country_id=obj.country_id)
        return obj

    @classmethod
    def _fetch_filter(
            cls, by_country_code=None, by_comment=None, by_ip=None,
            by_list=None):

        country_code_from_ip = None
        if by_ip:
            with geo.reader() as geo_reader:
                # country_code_from_ip will be None if here partial IP is provided  # noqa: E501
                country_code_from_ip = geo_reader.get_code(by_ip)

        q = CountryList.select(
            CountryList
        ).distinct().join(
            Country, JOIN.INNER, on=(CountryList.country == Country.id)
        )
        if by_list:
            q = q.where(CountryList.listname == by_list)
        q = q.order_by(Country.code)

        # filter by optional args
        if by_country_code:
            q = q.where(Country.code == by_country_code)
        if by_comment:
            q = q.where(cls.comment.contains(by_comment))
        if by_ip:
            q = q.where(Country.code == country_code_from_ip)
        return q

    @classmethod
    def fetch_count(cls, **filter_args):
        return cls._fetch_filter(**filter_args).count()

    @classmethod
    def fetch(cls, offset=None, limit=None, order_by=None, **filter_args):
        q = cls._fetch_filter(**filter_args)
        if offset is not None:
            q = q.offset(offset)
        if limit is not None:
            q = q.limit(limit)
        if order_by is not None:
            q = apply_order_by(order_by, cls, q)
        return [model_to_dict(row) for row in q]

    @classmethod
    def get_listname(cls, country):
        try:
            return cls.get(
                country=Country.select().where(Country.code == country)
            ).listname
        except DoesNotExist:
            return None

    @classmethod
    def delete_country(cls, country: Country, listname):
        deleted = CountryList.delete().where(
            (CountryList.country == country)
            & (CountryList.listname == listname)).execute()
        if deleted:
            cls.Signals.deleted.send(listname, country_id=country.id)
        return deleted

    @classmethod
    def country_codes(cls, listname) -> Generator:
        """Returns generator of listed country codes."""
        q = CountryList.select(
            Country.code
        ).distinct().join(
            Country, JOIN.INNER, on=(CountryList.country == Country.id)
        ).where(CountryList.listname == listname).order_by(
            Country.code
        )

        return (code for (code,) in q.tuples())
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: country.py

Server Info

System Features

User Info
