1
- # coding: utf-8
2
1
__version__ = "0.2.1"
3
2
4
3
import logging
7
6
import time
8
7
from decimal import Decimal
9
8
from enum import IntEnum
10
- from typing import Union , List , Dict , Literal
9
+ from typing import Dict , List , Literal , Union
11
10
12
- from netaddr import IPSet , IPNetwork
11
+ from netaddr import IPNetwork , IPSet
13
12
14
13
15
- class MmdbBaseType ( object ) :
14
+ class MmdbBaseType :
16
15
def __init__ (self , value ):
17
16
self .value = value
18
17
@@ -97,7 +96,7 @@ class MMDBTypeID(IntEnum):
97
96
UINT64_MAX = 0xFFFFFFFFFFFFFFFF
98
97
99
98
100
- class SearchTreeNode ( object ) :
99
+ class SearchTreeNode :
101
100
def __init__ (self , left = None , right = None ):
102
101
self .left = left
103
102
self .right = right
@@ -123,12 +122,12 @@ def __setitem__(self, key, value):
123
122
self .right = value
124
123
125
124
126
- class SearchTreeLeaf ( object ) :
125
+ class SearchTreeLeaf :
127
126
def __init__ (self , value ):
128
127
self .value = value
129
128
130
129
def __repr__ (self ):
131
- return "SearchTreeLeaf(value={value})" . format ( value = self .value )
130
+ return f "SearchTreeLeaf(value={ self .value } )"
132
131
133
132
__str__ = __repr__
134
133
@@ -156,7 +155,7 @@ def __repr__(self):
156
155
FloatType = Union [Literal ["f32" , "f64" , "float32" , "float64" ] | MmdbF32 | MmdbF64 ]
157
156
158
157
159
- class Encoder ( object ) :
158
+ class Encoder :
160
159
def __init__ (
161
160
self , cache = True , int_type : IntType = "auto" , float_type : FloatType = "f64"
162
161
):
@@ -224,7 +223,8 @@ def _encode_uint(self, type_id, max_len):
224
223
def _encode_unsigned_value (value ):
225
224
if value < 0 or value >= value_max :
226
225
raise ValueError (
227
- f"encode uint{ max_len * 8 } fail: { value } not in range(0, { value_max } )"
226
+ f"encode uint{ max_len * 8 } fail: "
227
+ f"{ value } not in range(0, { value_max } )"
228
228
)
229
229
res = b""
230
230
while value != 0 and len (res ) < max_len :
@@ -353,7 +353,7 @@ def python_type_id(self, value):
353
353
raise ValueError (f"unknown float_type={ self .float_type } " )
354
354
elif value_type is Decimal :
355
355
return MMDBTypeID .DOUBLE
356
- raise TypeError ("unknown type {value_type}" . format ( value_type = value_type ) )
356
+ raise TypeError (f "unknown type { value_type } " )
357
357
358
358
def encode_meta (self , meta ):
359
359
res = self ._make_header (MMDBTypeID .MAP , len (meta ))
@@ -383,8 +383,8 @@ def encode(self, value, type_id=None):
383
383
384
384
try :
385
385
encoder = self .type_encoder [type_id ]
386
- except KeyError :
387
- raise ValueError ("unknown type_id={type_id}" . format ( type_id = type_id ))
386
+ except KeyError as err :
387
+ raise ValueError (f "unknown type_id={ type_id } ") from err
388
388
389
389
if isinstance (value , MmdbBaseType ):
390
390
value = value .value
@@ -406,7 +406,7 @@ def encode(self, value, type_id=None):
406
406
return res
407
407
408
408
409
- class TreeWriter ( object ) :
409
+ class TreeWriter :
410
410
encoder_cls = Encoder
411
411
412
412
def __init__ (
@@ -541,7 +541,7 @@ def bits_rstrip(n, length=None, keep=0):
541
541
return map (int , bin (n )[2 :].rjust (length , "0" )[:keep ])
542
542
543
543
544
- class MMDBWriter ( object ) :
544
+ class MMDBWriter :
545
545
def __init__ (
546
546
self ,
547
547
ip_version = 4 ,
@@ -554,18 +554,20 @@ def __init__(
554
554
):
555
555
"""
556
556
Args:
557
- ip_version (int, optional) : The IP version of the database. Defaults to 4.
558
- database_type (str, optional) : The type of the database. Defaults to "GeoIP".
559
- languages (List[str], optional) : A list of languages. Defaults to [].
560
- description (Union[Dict[str, str], str], optional) : A description of the database for every language.
561
- ipv4_compatible (bool, optional) : Whether the database is compatible with IPv4. Defaults to False .
562
- int_type (Union[str, MmdbU16, MmdbU32, MmdbU64, MmdbU128, MmdbI32], optional) : The type of integer to use. Defaults to "auto".
563
- float_type (Union[str, MmdbF32, MmdbF64], optional) : The type of float to use. Defaults to "f64".
557
+ ip_version: The IP version of the database. Defaults to 4.
558
+ database_type: The type of the database. Defaults to "GeoIP".
559
+ languages: A list of languages. Defaults to [].
560
+ description: A description of the database for every language.
561
+ ipv4_compatible: Whether the database is compatible with IPv4.
562
+ int_type: The type of integer to use. Defaults to "auto".
563
+ float_type: The type of float to use. Defaults to "f64".
564
564
565
565
Note:
566
- If you want to store an IPv4 address in an IPv6 database, you should set ipv4_compatible=True.
566
+ If you want to store an IPv4 address in an IPv6 database, you should set
567
+ ipv4_compatible=True.
567
568
568
- If you want to use a specific integer type, you can set int_type to "u16", "u32", "u64", "u128", or "i32".
569
+ If you want to use a specific integer type, you can set int_type to
570
+ "u16", "u32", "u64", "u128", or "i32".
569
571
"""
570
572
self .tree = SearchTreeNode ()
571
573
self .ipv4_compatible = ipv4_compatible
@@ -582,16 +584,12 @@ def __init__(
582
584
self ._bit_length = 128 if ip_version == 6 else 32
583
585
584
586
if ip_version not in [4 , 6 ]:
585
- raise ValueError (
586
- "ip_version should be 4 or 6, {} is incorrect" .format (ip_version )
587
- )
587
+ raise ValueError (f"ip_version should be 4 or 6, { ip_version } is incorrect" )
588
588
if ip_version == 4 and ipv4_compatible :
589
589
raise ValueError ("ipv4_compatible=True can set when ip_version=6" )
590
590
if not self .binary_format_major_version :
591
591
raise ValueError (
592
- "major_version can't be empty or 0: {}" .format (
593
- self .binary_format_major_version
594
- )
592
+ f"major_version can't be empty or 0: { self .binary_format_major_version } "
595
593
)
596
594
if isinstance (description , str ):
597
595
self .description = {i : description for i in languages }
@@ -602,22 +600,22 @@ def __init__(
602
600
self .int_type = int_type
603
601
self .float_type = float_type
604
602
605
- def insert_network (
606
- self , network : IPSet , content : MMDBType , overwrite = True , python_type_id_map = None
607
- ):
603
+ def insert_network (self , network : IPSet , content : MMDBType ):
608
604
"""
609
605
Inserts a network into the MaxMind database.
610
606
611
607
Args:
612
- network (IPSet): The network to be inserted. It should be an instance of netaddr.IPSet.
613
- content (MMDBType): The content associated with the network. It can be a dictionary, list, string, bytes, integer, or boolean.
614
- overwrite (bool, optional): If True, existing network data will be overwritten. Defaults to True.
615
- python_type_id_map: abc
608
+ network: The network to be inserted. It should be an instance of
609
+ netaddr.IPSet.
610
+ content: The content associated with the network. It can be a
611
+ dictionary, list, string, bytes, integer, or boolean.
612
+
616
613
617
614
Raises:
618
615
ValueError: If the network is not an instance of netaddr.IPSet.
619
616
ValueError: If an IPv6 address is inserted into an IPv4-only database.
620
- ValueError: If an IPv4 address is inserted into an IPv6 database without setting ipv4_compatible=True.
617
+ ValueError: If an IPv4 address is inserted into an IPv6 database without
618
+ setting ipv4_compatible=True.
621
619
622
620
Note:
623
621
This method modifies the internal tree structure of the MMDBWriter instance.
@@ -629,15 +627,14 @@ def insert_network(
629
627
for cidr in network :
630
628
if self .ip_version == 4 and cidr .version == 6 :
631
629
raise ValueError (
632
- "You inserted a IPv6 address {} "
633
- "to an IPv4-only database." .format (cidr )
630
+ f"You inserted a IPv6 address { cidr } " "to an IPv4-only database."
634
631
)
635
632
if self .ip_version == 6 and cidr .version == 4 :
636
633
if not self .ipv4_compatible :
637
634
raise ValueError (
638
- "You inserted a IPv4 address {} to an IPv6 database."
635
+ f "You inserted a IPv4 address { cidr } to an IPv6 database."
639
636
"Please use ipv4_compatible=True option store "
640
- "IPv4 address in IPv6 database as ::/96 format" . format ( cidr )
637
+ "IPv4 address in IPv6 database as ::/96 format"
641
638
)
642
639
cidr = cidr .ipv6 (True )
643
640
node = self .tree
@@ -661,15 +658,17 @@ def insert_network(
661
658
)
662
659
)
663
660
logger .info (
664
- f"Inserting { cidr } ({ content } ) into subnet of { current_cidr } ({ current_node .value } )"
661
+ f"Inserting { cidr } ({ content } ) into subnet of "
662
+ f"{ current_cidr } ({ current_node .value } )"
665
663
)
666
664
supernet_leaf = current_node
667
665
current_node = SearchTreeNode ()
668
666
previous_node [ip_bit ] = current_node
669
667
670
668
if supernet_leaf :
671
669
next_bit = bits [index + 1 ]
672
- # Insert supernet information on each inverse bit of the current subnet
670
+ # Insert supernet information on each inverse bit of
671
+ # the current subnet
673
672
current_node [1 - next_bit ] = supernet_leaf
674
673
current_node [bits [- 1 ]] = leaf
675
674
0 commit comments