fix: lint

Author: Nekotoxin

casbin_adapter/adapter.py

@@ -53,22 +53,22 @@ def add_policy(self, sec, ptype, rule):
 
     def remove_policy(self, sec, ptype, rule):
         """removes a policy rule from the storage."""
-        query_params = {'ptype': ptype}
+        query_params = {"ptype": ptype}
         for i, v in enumerate(rule):
-            query_params['v{}'.format(i)] = v
+            query_params["v{}".format(i)] = v
         rows_deleted, _ = CasbinRule.objects.filter(**query_params).delete()
         return True if rows_deleted > 0 else False
 
     def remove_filtered_policy(self, sec, ptype, field_index, *field_values):
         """removes policy rules that match the filter from the storage.
         This is part of the Auto-Save feature.
         """
-        query_params = {'ptype': ptype}
-        if not(0 <= field_index <= 5):
+        query_params = {"ptype": ptype}
+        if not (0 <= field_index <= 5):
             return False
         if not (1 <= field_index + len(field_values) <= 6):
             return False
         for i, v in enumerate(field_values):
-            query_params['v{}'.format(i + field_index)] = v
+            query_params["v{}".format(i + field_index)] = v
         rows_deleted, _ = CasbinRule.objects.filter(**query_params).delete()
         return True if rows_deleted > 0 else False

casbin_adapter/apps.py

@@ -4,9 +4,9 @@
 
 
 class CasbinAdapterConfig(AppConfig):
-    name = 'casbin_adapter'
+    name = "casbin_adapter"
 
     def ready(self):
         from .enforcer import initialize_enforcer
-        initialize_enforcer()
 
+        initialize_enforcer()

casbin_adapter/enforcer.py

@@ -11,47 +11,48 @@
 logger = logging.getLogger(__name__)
 
 
-
 class ProxyEnforcer(Enforcer):
     _initialized = False
 
     def __init__(self, *args, **kwargs):
         if self._initialized:
             super().__init__(*args, **kwargs)
         else:
-            logger.info('Deferring casbin enforcer initialisation until django is ready')
+            logger.info("Deferring casbin enforcer initialisation until django is ready")
 
     def _load(self):
         if self._initialized == False:
-            logger.info('Performing deferred casbin enforcer initialisation')
+            logger.info("Performing deferred casbin enforcer initialisation")
             self._initialized = True
-            model = getattr(settings, 'CASBIN_MODEL')
-            enable_log = getattr(settings, 'CASBIN_LOG_ENABLED', False)
-            adapter_loc = getattr(settings, 'CASBIN_ADAPTER', 'casbin_adapter.adapter.Adapter')
-            adapter_args = getattr(settings, 'CASBIN_ADAPTER_ARGS', tuple())
+            model = getattr(settings, "CASBIN_MODEL")
+            enable_log = getattr(settings, "CASBIN_LOG_ENABLED", False)
+            adapter_loc = getattr(settings, "CASBIN_ADAPTER", "casbin_adapter.adapter.Adapter")
+            adapter_args = getattr(settings, "CASBIN_ADAPTER_ARGS", tuple())
             Adapter = import_class(adapter_loc)
             adapter = Adapter(*adapter_args)
 
             super().__init__(model, adapter, enable_log)
-            logger.debug('Casbin enforcer initialised')
+            logger.debug("Casbin enforcer initialised")
 
-            watcher = getattr(settings, 'CASBIN_WATCHER', None)
+            watcher = getattr(settings, "CASBIN_WATCHER", None)
             if watcher:
                 self.set_watcher(watcher)
 
-            role_manager = getattr(settings, 'CASBIN_ROLE_MANAGER', None)
+            role_manager = getattr(settings, "CASBIN_ROLE_MANAGER", None)
             if role_manager:
                 self.set_role_manager(role_manager)
 
     def __getattribute__(self, name):
-        safe_methods = ['__init__', '_load', '_initialized']
-        if not super().__getattribute__('_initialized') and name not in safe_methods:
+        safe_methods = ["__init__", "_load", "_initialized"]
+        if not super().__getattribute__("_initialized") and name not in safe_methods:
             initialize_enforcer()
-            if not super().__getattribute__('_initialized'):
-                raise Exception((
-                    "Calling enforcer attributes before django registry is ready. "
-                    "Prevent making any calls to the enforcer on import/startup"
-                ))
+            if not super().__getattribute__("_initialized"):
+                raise Exception(
+                    (
+                        "Calling enforcer attributes before django registry is ready. "
+                        "Prevent making any calls to the enforcer on import/startup"
+                    )
+                )
 
         return super().__getattribute__(name)
 
@@ -73,4 +74,3 @@ def initialize_enforcer():
                 enforcer._load()
     except (OperationalError, ProgrammingError):
         pass
-

casbin_adapter/migrations/0001_initial.py

@@ -7,24 +7,23 @@ class Migration(migrations.Migration):
 
     initial = True
 
-    dependencies = [
-    ]
+    dependencies = []
 
     operations = [
         migrations.CreateModel(
-            name='CasbinRule',
+            name="CasbinRule",
             fields=[
-                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
-                ('ptype', models.CharField(max_length=255)),
-                ('v0', models.CharField(max_length=255)),
-                ('v1', models.CharField(max_length=255)),
-                ('v2', models.CharField(max_length=255)),
-                ('v3', models.CharField(max_length=255)),
-                ('v4', models.CharField(max_length=255)),
-                ('v5', models.CharField(max_length=255)),
+                ("id", models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("ptype", models.CharField(max_length=255)),
+                ("v0", models.CharField(max_length=255)),
+                ("v1", models.CharField(max_length=255)),
+                ("v2", models.CharField(max_length=255)),
+                ("v3", models.CharField(max_length=255)),
+                ("v4", models.CharField(max_length=255)),
+                ("v5", models.CharField(max_length=255)),
             ],
             options={
-                'db_table': 'casbin_rule',
+                "db_table": "casbin_rule",
             },
         ),
     ]

casbin_adapter/models.py

@@ -13,23 +13,23 @@ class CasbinRule(models.Model):
     v5 = models.CharField(max_length=255)
 
     class Meta:
-        db_table = 'casbin_rule'
+        db_table = "casbin_rule"
 
     def __str__(self):
         text = self.ptype
 
         if self.v0:
-            text = text + ', ' + self.v0
+            text = text + ", " + self.v0
         if self.v1:
-            text = text + ', ' + self.v1
+            text = text + ", " + self.v1
         if self.v2:
-            text = text + ', ' + self.v2
+            text = text + ", " + self.v2
         if self.v3:
-            text = text + ', ' + self.v3
+            text = text + ", " + self.v3
         if self.v4:
-            text = text + ', ' + self.v4
+            text = text + ", " + self.v4
         if self.v5:
-            text = text + ', ' + self.v5
+            text = text + ", " + self.v5
         return text
 
     def __repr__(self):

casbin_adapter/utils.py

@@ -1,10 +1,11 @@
 import importlib
 
+
 def import_class(name):
     """Import class from string
     e.g. `package.module.ClassToImport` returns the `ClasToImport` class"""
-    components = name.split('.')
-    module_name = '.'.join(components[:-1])
+    components = name.split(".")
+    module_name = ".".join(components[:-1])
     class_name = components[-1]
     module = importlib.import_module(module_name)
     class_ = getattr(module, class_name)

setup.py

@@ -27,4 +27,4 @@
         "Operating System :: OS Independent",
     ],
     data_files=[desc_file],
-)
+)

tests/settings.py

@@ -2,7 +2,7 @@
 
 BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 
-SECRET_KEY = 'not-a-production-secret'
+SECRET_KEY = "not-a-production-secret"
 
 INSTALLED_APPS = [
     "django.contrib.admin",
@@ -42,4 +42,4 @@
 
 DATABASES = {"default": {"ENGINE": "django.db.backends.sqlite3", "NAME": ":memory:"}}
 
-CASBIN_MODEL = os.path.join(BASE_DIR, 'tests', 'rbac_model.conf')
+CASBIN_MODEL = os.path.join(BASE_DIR, "tests", "rbac_model.conf")

tests/test_adapter.py

@@ -12,117 +12,119 @@ def get_fixture(path):
     dir_path = os.path.split(os.path.realpath(__file__))[0] + "/"
     return os.path.abspath(dir_path + path)
 
+
 def get_enforcer():
     adapter = Adapter()
 
-    CasbinRule.objects.bulk_create([
-        CasbinRule(ptype='p', v0='alice', v1='data1', v2='read'),
-        CasbinRule(ptype='p', v0='bob', v1='data2', v2='write'),
-        CasbinRule(ptype='p', v0='data2_admin', v1='data2', v2='read'),
-        CasbinRule(ptype='p', v0='data2_admin', v1='data2', v2='write'),
-        CasbinRule(ptype='g', v0='alice', v1='data2_admin'),
-    ])
+    CasbinRule.objects.bulk_create(
+        [
+            CasbinRule(ptype="p", v0="alice", v1="data1", v2="read"),
+            CasbinRule(ptype="p", v0="bob", v1="data2", v2="write"),
+            CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="read"),
+            CasbinRule(ptype="p", v0="data2_admin", v1="data2", v2="write"),
+            CasbinRule(ptype="g", v0="alice", v1="data2_admin"),
+        ]
+    )
 
-    return casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
+    return casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
 
 
 class TestConfig(TestCase):
     def test_enforcer_basic(self):
         e = get_enforcer()
-        self.assertTrue(e.enforce('alice', 'data1', 'read'))
-        self.assertFalse(e.enforce('bob', 'data1', 'read'))
-        self.assertTrue(e.enforce('bob', 'data2', 'write'))
-        self.assertTrue(e.enforce('alice', 'data2', 'read'))
-        self.assertTrue(e.enforce('alice', 'data2', 'write'))
+        self.assertTrue(e.enforce("alice", "data1", "read"))
+        self.assertFalse(e.enforce("bob", "data1", "read"))
+        self.assertTrue(e.enforce("bob", "data2", "write"))
+        self.assertTrue(e.enforce("alice", "data2", "read"))
+        self.assertTrue(e.enforce("alice", "data2", "write"))
 
     def test_add_policy(self):
         adapter = Adapter()
-        e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
+        e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
 
         try:
-            self.assertFalse(e.enforce('alice', 'data1', 'read'))
-            self.assertFalse(e.enforce('bob', 'data1', 'read'))
-            self.assertFalse(e.enforce('bob', 'data2', 'write'))
-            self.assertFalse(e.enforce('alice', 'data2', 'read'))
-            self.assertFalse(e.enforce('alice', 'data2', 'write'))
+            self.assertFalse(e.enforce("alice", "data1", "read"))
+            self.assertFalse(e.enforce("bob", "data1", "read"))
+            self.assertFalse(e.enforce("bob", "data2", "write"))
+            self.assertFalse(e.enforce("alice", "data2", "read"))
+            self.assertFalse(e.enforce("alice", "data2", "write"))
         except simpleeval.NameNotDefined:
             # This is caused by an upstream bug when there is no policy loaded
             # Should be resolved in pycasbin >= 0.3
             pass
 
-        adapter.add_policy(sec=None, ptype='p', rule=['alice', 'data1', 'read'])
-        adapter.add_policy(sec=None, ptype='p', rule=['bob', 'data2', 'write'])
-        adapter.add_policy(sec=None, ptype='p', rule=['data2_admin', 'data2', 'read'])
-        adapter.add_policy(sec=None, ptype='p', rule=['data2_admin', 'data2', 'write'])
-        adapter.add_policy(sec=None, ptype='g', rule=['alice', 'data2_admin'])
+        adapter.add_policy(sec=None, ptype="p", rule=["alice", "data1", "read"])
+        adapter.add_policy(sec=None, ptype="p", rule=["bob", "data2", "write"])
+        adapter.add_policy(sec=None, ptype="p", rule=["data2_admin", "data2", "read"])
+        adapter.add_policy(sec=None, ptype="p", rule=["data2_admin", "data2", "write"])
+        adapter.add_policy(sec=None, ptype="g", rule=["alice", "data2_admin"])
 
         e.load_policy()
 
-        self.assertTrue(e.enforce('alice', 'data1', 'read'))
-        self.assertFalse(e.enforce('bob', 'data1', 'read'))
-        self.assertTrue(e.enforce('bob', 'data2', 'write'))
-        self.assertTrue(e.enforce('alice', 'data2', 'read'))
-        self.assertTrue(e.enforce('alice', 'data2', 'write'))
-        self.assertFalse(e.enforce('bogus', 'data2', 'write'))
+        self.assertTrue(e.enforce("alice", "data1", "read"))
+        self.assertFalse(e.enforce("bob", "data1", "read"))
+        self.assertTrue(e.enforce("bob", "data2", "write"))
+        self.assertTrue(e.enforce("alice", "data2", "read"))
+        self.assertTrue(e.enforce("alice", "data2", "write"))
+        self.assertFalse(e.enforce("bogus", "data2", "write"))
 
     def test_save_policy(self):
-        model = casbin.Enforcer(get_fixture('rbac_model.conf'), get_fixture('rbac_policy.csv')).model
+        model = casbin.Enforcer(get_fixture("rbac_model.conf"), get_fixture("rbac_policy.csv")).model
         adapter = Adapter()
         adapter.save_policy(model)
-        e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
+        e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
 
-        self.assertTrue(e.enforce('alice', 'data1', 'read'))
-        self.assertFalse(e.enforce('bob', 'data1', 'read'))
-        self.assertTrue(e.enforce('bob', 'data2', 'write'))
-        self.assertTrue(e.enforce('alice', 'data2', 'read'))
-        self.assertTrue(e.enforce('alice', 'data2', 'write'))
+        self.assertTrue(e.enforce("alice", "data1", "read"))
+        self.assertFalse(e.enforce("bob", "data1", "read"))
+        self.assertTrue(e.enforce("bob", "data2", "write"))
+        self.assertTrue(e.enforce("alice", "data2", "read"))
+        self.assertTrue(e.enforce("alice", "data2", "write"))
 
     def test_autosave_off_doesnt_persist_to_db(self):
         adapter = Adapter()
-        e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
+        e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
 
         e.enable_auto_save(False)
-        e.add_policy('alice', 'data1', 'write')
+        e.add_policy("alice", "data1", "write")
         e.load_policy()
         policies = e.get_policy()
         self.assertListEqual(policies, [])
 
     def test_autosave_on_persists_to_db(self):
         adapter = Adapter()
-        e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
+        e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
 
         e.enable_auto_save(True)
-        e.add_policy('alice', 'data1', 'write')
+        e.add_policy("alice", "data1", "write")
         e.load_policy()
         policies = e.get_policy()
-        self.assertListEqual(policies, [['alice', 'data1', 'write']])
+        self.assertListEqual(policies, [["alice", "data1", "write"]])
 
     def test_autosave_on_persists_remove_action_to_db(self):
         adapter = Adapter()
-        e = casbin.Enforcer(get_fixture('rbac_model.conf'), adapter)
-        e.add_policy('alice', 'data1', 'write')
+        e = casbin.Enforcer(get_fixture("rbac_model.conf"), adapter)
+        e.add_policy("alice", "data1", "write")
         e.load_policy()
-        self.assertListEqual(e.get_policy(), [['alice', 'data1', 'write']])
+        self.assertListEqual(e.get_policy(), [["alice", "data1", "write"]])
 
         e.enable_auto_save(True)
-        e.remove_policy('alice', 'data1', 'write')
+        e.remove_policy("alice", "data1", "write")
         e.load_policy()
         self.assertListEqual(e.get_policy(), [])
 
     def test_remove_filtered_policy(self):
         e = get_enforcer()
 
-        e.remove_filtered_policy(0, 'data2_admin')
+        e.remove_filtered_policy(0, "data2_admin")
         e.load_policy()
-        self.assertListEqual(e.get_policy(), [['alice', 'data1', 'read'], ['bob', 'data2', 'write']])
-
+        self.assertListEqual(e.get_policy(), [["alice", "data1", "read"], ["bob", "data2", "write"]])
 
     def test_str(self):
-        rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read')
-        self.assertEqual(str(rule), 'p, alice, data1, read')
+        rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read")
+        self.assertEqual(str(rule), "p, alice, data1, read")
 
     def test_repr(self):
-        rule = CasbinRule(ptype='p', v0='alice', v1='data1', v2='read')
+        rule = CasbinRule(ptype="p", v0="alice", v1="data1", v2="read")
         self.assertEqual(repr(rule), '<CasbinRule None: "p, alice, data1, read">')
         rule.save()
         self.assertRegex(repr(rule), r'<CasbinRule \d+: "p, alice, data1, read">')