2282901844

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/pycasbin/sqlalchemy-adapter/commit/<a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/<a class="double-link" href="https://git"><a class=hub.com/pycasbin/sqlalchemy-adapter/commit/7db6b59b54a1d63d0fd602d5d2f9846f3807f16f">7db6b59b5">&lt;a href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/</a><a class="double-link" href="https://github.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/&lt;a class=&quot;double-link&quot; href=&quot;https://git">&lt;a class=</a>hub.com/pycasbin/sqlalchemy-adapter/commit/7db6b59b54a1d63d0fd602d5d2f9846f3807f16f">7db6b59b5</a><a href="https://github.com/pycasbin/sqlalchemy-adapter/commit/7db6b59b54a1d63d0fd602d5d2f9846f3807f16f">&quot;&gt;&amp;lt;a href=&amp;quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/&lt;/a&gt;&lt;a class=&quot;double-link&quot; href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/pycasbin/sqlalchemy-adapter/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/pycasbin/sqlalchemy-adapter/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/pycasbin/sqlalchemy-adapter/commit/&amp;lt;a class=&amp;quot;double-link&amp;quot; href=&amp;quot;https://git&quot;&gt;&amp;lt;a class=&lt;/a&gt;hub.com/pycasbin/sqlalchemy-adapter/commit/7db6b59b54a1d63d0fd602d5d2f9846f3807f16f&quot;&gt;7db6b59b5&lt;/a&gt;&lt;a href=&quot;https://github.com/pycasbin/sqlalchemy-adapter/commit/7db6b59b54a1d63d0fd602d5d2f9846f3807f16f&quot;&gt;&amp;quot;&amp;gt;&amp;amp;lt;a href=&amp;amp;quot;https://github.com/pycasbin/sqlal... (continued)

Pull Request Pull Request #56: feat: update_filtered_policies

Run Details

2 of 16 new or added lines in 1 file covered. (12.5%)

139 of 163 relevant lines covered (85.28%)

3.41 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.19

/casbin_sqlalchemy_adapter/adapter.py

from contextlib import contextmanager

from casbin import persist
from sqlalchemy import Column, Integer, String
from sqlalchemy import create_engine, or_
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()


class CasbinRule(Base):
    __tablename__ = "casbin_rule"

    id = Column(Integer, primary_key=True)
    ptype = Column(String(255))
    v0 = Column(String(255))
    v1 = Column(String(255))
    v2 = Column(String(255))
    v3 = Column(String(255))
    v4 = Column(String(255))
    v5 = Column(String(255))

    def __str__(self):
        arr = [self.ptype]
        for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5):
            if v is None:
                break
            arr.append(v)
        return ", ".join(arr)

    def __repr__(self):
        return '<CasbinRule {}: "{}">'.format(self.id, str(self))


class Filter:
    ptype = []
    v0 = []
    v1 = []
    v2 = []
    v3 = []
    v4 = []
    v5 = []


class Adapter(persist.Adapter, persist.adapters.UpdateAdapter):
    """the interface for Casbin adapters."""

    def __init__(self, engine, db_class=None, filtered=False):
        if isinstance(engine, str):
            self._engine = create_engine(engine)
        else:
            self._engine = engine

        if db_class is None:
            db_class = CasbinRule
        else:
            for attr in (
                "id",
                "ptype",
                "v0",
                "v1",
                "v2",
                "v3",
                "v4",
                "v5",
            ):  # id attr was used by filter
                if not hasattr(db_class, attr):
                    raise Exception(f"{attr} not found in custom DatabaseClass.")
            Base.metadata = db_class.metadata

        self._db_class = db_class
        self.session_local = sessionmaker(bind=self._engine)

        Base.metadata.create_all(self._engine)
        self._filtered = filtered

    @contextmanager
    def _session_scope(self):
        """Provide a transactional scope around a series of operations."""
        session = self.session_local()
        try:
            yield session
            session.commit()
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()

    def load_policy(self, model):
        """loads all policy rules from the storage."""
        with self._session_scope() as session:
            lines = session.query(self._db_class).all()
            for line in lines:
                persist.load_policy_line(str(line), model)

    def is_filtered(self):
        return self._filtered

    def load_filtered_policy(self, model, filter) -> None:
        """loads all policy rules from the storage."""
        with self._session_scope() as session:
            query = session.query(self._db_class)
            filters = self.filter_query(query, filter)
            filters = filters.all()

            for line in filters:
                persist.load_policy_line(str(line), model)
            self._filtered = True

    def filter_query(self, querydb, filter):
        for attr in ("ptype", "v0", "v1", "v2", "v3", "v4", "v5"):
            if len(getattr(filter, attr)) > 0:
                querydb = querydb.filter(
                    getattr(self._db_class, attr).in_(getattr(filter, attr))
                )
        return querydb.order_by(self._db_class.id)

    def _save_policy_line(self, ptype, rule):
        with self._session_scope() as session:
            line = self._db_class(ptype=ptype)
            for i, v in enumerate(rule):
                setattr(line, "v{}".format(i), v)
            session.add(line)

    def save_policy(self, model):
        """saves all policy rules to the storage."""
        with self._session_scope() as session:
            query = session.query(self._db_class)
            query.delete()
            for sec in ["p", "g"]:
                if sec not in model.model.keys():
                    continue
                for ptype, ast in model.model[sec].items():
                    for rule in ast.policy:
                        self._save_policy_line(ptype, rule)
        return True

    def add_policy(self, sec, ptype, rule):
        """adds a policy rule to the storage."""
        self._save_policy_line(ptype, rule)

    def add_policies(self, sec, ptype, rules):
        """adds a policy rules to the storage."""
        for rule in rules:
            self._save_policy_line(ptype, rule)

    def remove_policy(self, sec, ptype, rule):
        """removes a policy rule from the storage."""
        with self._session_scope() as session:
            query = session.query(self._db_class)
            query = query.filter(self._db_class.ptype == ptype)
            for i, v in enumerate(rule):
                query = query.filter(getattr(self._db_class, "v{}".format(i)) == v)
            r = query.delete()

        return True if r > 0 else False

    def remove_policies(self, sec, ptype, rules):
        """remove policy rules from the storage."""
        if not rules:
            return
        with self._session_scope() as session:
            query = session.query(self._db_class)
            query = query.filter(self._db_class.ptype == ptype)
            rules = zip(*rules)
            for i, rule in enumerate(rules):
                query = query.filter(
                    or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)
                )
            query.delete()

    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.
        """
        with self._session_scope() as session:
            query = session.query(self._db_class).filter(self._db_class.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):
                if v != "":
                    v_value = getattr(self._db_class, "v{}".format(field_index + i))
                    query = query.filter(v_value == v)
            r = query.delete()

        return True if r > 0 else False

    def update_policy(
        self, sec: str, ptype: str, old_rule: [str], new_rule: [str]
    ) -> None:
        """
        Update the old_rule with the new_rule in the database (storage).

        :param sec: section type
        :param ptype: policy type
        :param old_rule: the old rule that needs to be modified
        :param new_rule: the new rule to replace the old rule

        :return: None
        """

        with self._session_scope() as session:
            query = session.query(self._db_class).filter(self._db_class.ptype == ptype)

            # locate the old rule
            for index, value in enumerate(old_rule):
                v_value = getattr(self._db_class, "v{}".format(index))
                query = query.filter(v_value == value)

            # need the length of the longest_rule to perform overwrite
            longest_rule = old_rule if len(old_rule) > len(new_rule) else new_rule
            old_rule_line = query.one()

            # overwrite the old rule with the new rule
            for index in range(len(longest_rule)):
                if index < len(new_rule):
                    exec(f"old_rule_line.v{index} = new_rule[{index}]")
                else:
                    exec(f"old_rule_line.v{index} = None")

    def update_policies(
        self,
        sec: str,
        ptype: str,
        old_rules: [
            [str],
        ],
        new_rules: [
            [str],
        ],
    ) -> None:
        """
        Update the old_rules with the new_rules in the database (storage).

        :param sec: section type
        :param ptype: policy type
        :param old_rules: the old rules that need to be modified
        :param new_rules: the new rules to replace the old rules

        :return: None
        """
        for i in range(len(old_rules)):
            self.update_policy(sec, ptype, old_rules[i], new_rules[i])

    def update_filtered_policies(self, sec, ptype, new_rules: [[str]], field_index, *field_values) -> [[str]]:
        """update_filtered_policies updates all the policies on the basis of the filter."""

        filter = Filter()
        filter.ptype = ptype

        # Creating Filter from the field_index & field_values provided
        for i in range(len(field_values)):
            if field_index <= i and i < field_index+len(field_values):
                setattr(filter, f"v{i}", field_values[i-field_index])
            else:
                break
        
        self._update_filtered_policies(new_rules, filter)
    
    def _update_filtered_policies(self,  new_rules, filter) -> [[str]]:
        """_update_filtered_policies updates all the policies on the basis of the filter."""

        with self._session_scope() as session:

            # Load old policies

            query = session.query(self._db_class).filter(self._db_class.ptype == filter.ptype)
            filtered_query = self.filter_query(query, filter)
            old_rules = filtered_query.all()

            # Delete old policies

            self.remove_policies("p", filter.ptype, old_rules)

            # Insert new policies

            self.add_policies("p", filter.ptype, new_rules)

            # return deleted rules

            return old_rules

1	from contextlib import contextmanager	4✔
2
3	from casbin import persist	4✔
4	from sqlalchemy import Column, Integer, String	4✔
5	from sqlalchemy import create_engine, or_	4✔
6	from sqlalchemy.ext.declarative import declarative_base	4✔
7	from sqlalchemy.orm import sessionmaker	4✔
8
9	Base = declarative_base()	4✔
10
11
12	class CasbinRule(Base):	4✔
13	__tablename__ = "casbin_rule"	4✔
14
15	id = Column(Integer, primary_key=True)	4✔
16	ptype = Column(String(255))	4✔
17	v0 = Column(String(255))	4✔
18	v1 = Column(String(255))	4✔
19	v2 = Column(String(255))	4✔
20	v3 = Column(String(255))	4✔
21	v4 = Column(String(255))	4✔
22	v5 = Column(String(255))	4✔
23
24	def __str__(self):	4✔
25	arr = [self.ptype]	4✔
26	for v in (self.v0, self.v1, self.v2, self.v3, self.v4, self.v5):	4✔
27	if v is None:	4✔
28	break	4✔
29	arr.append(v)	4✔
30	return ", ".join(arr)	4✔
31
32	def __repr__(self):	4✔
33	return '<CasbinRule {}: "{}">'.format(self.id, str(self))	4✔
34
35
36	class Filter:	4✔
37	ptype = []	4✔
38	v0 = []	4✔
39	v1 = []	4✔
40	v2 = []	4✔
41	v3 = []	4✔
42	v4 = []	4✔
43	v5 = []	4✔
44
45
46	class Adapter(persist.Adapter, persist.adapters.UpdateAdapter):	4✔
47	"""the interface for Casbin adapters."""
48
49	def __init__(self, engine, db_class=None, filtered=False):	4✔
50	if isinstance(engine, str):	4✔
51	self._engine = create_engine(engine)	×
52	else:
53	self._engine = engine	4✔
54
55	if db_class is None:	4✔
56	db_class = CasbinRule	4✔
57	else:
58	for attr in (	4✔
59	"id",
60	"ptype",
61	"v0",
62	"v1",
63	"v2",
64	"v3",
65	"v4",
66	"v5",
67	): # id attr was used by filter
68	if not hasattr(db_class, attr):	4✔
69	raise Exception(f"{attr} not found in custom DatabaseClass.")	×
70	Base.metadata = db_class.metadata	4✔
71
72	self._db_class = db_class	4✔
73	self.session_local = sessionmaker(bind=self._engine)	4✔
74
75	Base.metadata.create_all(self._engine)	4✔
76	self._filtered = filtered	4✔
77
78	@contextmanager	4✔
79	def _session_scope(self):	2✔
80	"""Provide a transactional scope around a series of operations."""
81	session = self.session_local()	4✔
82	try:	4✔
83	yield session	4✔
84	session.commit()	4✔
85	except Exception as e:	×
86	session.rollback()	×
87	raise e	×
88	finally:
89	session.close()	4✔
90
91	def load_policy(self, model):	4✔
92	"""loads all policy rules from the storage."""
93	with self._session_scope() as session:	4✔
94	lines = session.query(self._db_class).all()	4✔
95	for line in lines:	4✔
96	persist.load_policy_line(str(line), model)	4✔
97
98	def is_filtered(self):	4✔
99	return self._filtered	4✔
100
101	def load_filtered_policy(self, model, filter) -> None:	4✔
102	"""loads all policy rules from the storage."""
103	with self._session_scope() as session:	4✔
104	query = session.query(self._db_class)	4✔
105	filters = self.filter_query(query, filter)	4✔
106	filters = filters.all()	4✔
107
108	for line in filters:	4✔
109	persist.load_policy_line(str(line), model)	4✔
110	self._filtered = True	4✔
111
112	def filter_query(self, querydb, filter):	4✔
113	for attr in ("ptype", "v0", "v1", "v2", "v3", "v4", "v5"):	4✔
114	if len(getattr(filter, attr)) > 0:	4✔
115	querydb = querydb.filter(	4✔
116	getattr(self._db_class, attr).in_(getattr(filter, attr))
117	)
118	return querydb.order_by(self._db_class.id)	4✔
119
120	def _save_policy_line(self, ptype, rule):	4✔
121	with self._session_scope() as session:	4✔
122	line = self._db_class(ptype=ptype)	4✔
123	for i, v in enumerate(rule):	4✔
124	setattr(line, "v{}".format(i), v)	4✔
125	session.add(line)	4✔
126
127	def save_policy(self, model):	4✔
128	"""saves all policy rules to the storage."""
129	with self._session_scope() as session:	4✔
130	query = session.query(self._db_class)	4✔
131	query.delete()	4✔
132	for sec in ["p", "g"]:	4✔
133	if sec not in model.model.keys():	4✔
134	continue	×
135	for ptype, ast in model.model[sec].items():	4✔
136	for rule in ast.policy:	4✔
137	self._save_policy_line(ptype, rule)	4✔
138	return True	4✔
139
140	def add_policy(self, sec, ptype, rule):	4✔
141	"""adds a policy rule to the storage."""
142	self._save_policy_line(ptype, rule)	4✔
143
144	def add_policies(self, sec, ptype, rules):	4✔
145	"""adds a policy rules to the storage."""
146	for rule in rules:	4✔
147	self._save_policy_line(ptype, rule)	4✔
148
149	def remove_policy(self, sec, ptype, rule):	4✔
150	"""removes a policy rule from the storage."""
151	with self._session_scope() as session:	4✔
152	query = session.query(self._db_class)	4✔
153	query = query.filter(self._db_class.ptype == ptype)	4✔
154	for i, v in enumerate(rule):	4✔
155	query = query.filter(getattr(self._db_class, "v{}".format(i)) == v)	4✔
156	r = query.delete()	4✔
157
158	return True if r > 0 else False	4✔
159
160	def remove_policies(self, sec, ptype, rules):	4✔
161	"""remove policy rules from the storage."""
162	if not rules:	4✔
163	return	×
164	with self._session_scope() as session:	4✔
165	query = session.query(self._db_class)	4✔
166	query = query.filter(self._db_class.ptype == ptype)	4✔
167	rules = zip(*rules)	4✔
168	for i, rule in enumerate(rules):	4✔
169	query = query.filter(	4✔
170	or_(getattr(self._db_class, "v{}".format(i)) == v for v in rule)
171	)
172	query.delete()	4✔
173
174	def remove_filtered_policy(self, sec, ptype, field_index, *field_values):	4✔
175	"""removes policy rules that match the filter from the storage.
176	This is part of the Auto-Save feature.
177	"""
178	with self._session_scope() as session:	4✔
179	query = session.query(self._db_class).filter(self._db_class.ptype == ptype)	4✔
180
181	if not (0 <= field_index <= 5):	4✔
182	return False	×
183	if not (1 <= field_index + len(field_values) <= 6):	4✔
184	return False	×
185	for i, v in enumerate(field_values):	4✔
186	if v != "":	4✔
187	v_value = getattr(self._db_class, "v{}".format(field_index + i))	4✔
188	query = query.filter(v_value == v)	4✔
189	r = query.delete()	4✔
190
191	return True if r > 0 else False	4✔
192
193	def update_policy(	4✔
194	self, sec: str, ptype: str, old_rule: [str], new_rule: [str]
195	) -> None:
196	"""
197	Update the old_rule with the new_rule in the database (storage).
198
199	:param sec: section type
200	:param ptype: policy type
201	:param old_rule: the old rule that needs to be modified
202	:param new_rule: the new rule to replace the old rule
203
204	:return: None
205	"""
206
207	with self._session_scope() as session:	4✔
208	query = session.query(self._db_class).filter(self._db_class.ptype == ptype)	4✔
209
210	# locate the old rule
211	for index, value in enumerate(old_rule):	4✔
212	v_value = getattr(self._db_class, "v{}".format(index))	4✔
213	query = query.filter(v_value == value)	4✔
214
215	# need the length of the longest_rule to perform overwrite
216	longest_rule = old_rule if len(old_rule) > len(new_rule) else new_rule	4✔
217	old_rule_line = query.one()	4✔
218
219	# overwrite the old rule with the new rule
220	for index in range(len(longest_rule)):	4✔
221	if index < len(new_rule):	4✔
222	exec(f"old_rule_line.v{index} = new_rule[{index}]")	4✔
223	else:
224	exec(f"old_rule_line.v{index} = None")	×
225
226	def update_policies(	4✔
227	self,
228	sec: str,
229	ptype: str,
230	old_rules: [
231	[str],
232	],
233	new_rules: [
234	[str],
235	],
236	) -> None:
237	"""
238	Update the old_rules with the new_rules in the database (storage).
239
240	:param sec: section type
241	:param ptype: policy type
242	:param old_rules: the old rules that need to be modified
243	:param new_rules: the new rules to replace the old rules
244
245	:return: None
246	"""
247	for i in range(len(old_rules)):	4✔
248	self.update_policy(sec, ptype, old_rules[i], new_rules[i])	4✔
249
250	def update_filtered_policies(self, sec, ptype, new_rules: [[str]], field_index, *field_values) -> [[str]]:	4✔
251	"""update_filtered_policies updates all the policies on the basis of the filter."""
252
NEW 253	filter = Filter()	×
NEW 254	filter.ptype = ptype	×
255
256	# Creating Filter from the field_index & field_values provided
NEW 257	for i in range(len(field_values)):	×
NEW 258	if field_index <= i and i < field_index+len(field_values):	×
NEW 259	setattr(filter, f"v{i}", field_values[i-field_index])	×
260	else:
NEW 261	break	×
262
NEW 263	self._update_filtered_policies(new_rules, filter)	×
264
265	def _update_filtered_policies(self, new_rules, filter) -> [[str]]:	4✔
266	"""_update_filtered_policies updates all the policies on the basis of the filter."""
267
NEW 268	with self._session_scope() as session:	×
269
270	# Load old policies
271
NEW 272	query = session.query(self._db_class).filter(self._db_class.ptype == filter.ptype)	×
NEW 273	filtered_query = self.filter_query(query, filter)	×
NEW 274	old_rules = filtered_query.all()	×
275
276	# Delete old policies
277
NEW 278	self.remove_policies("p", filter.ptype, old_rules)	×
279
280	# Insert new policies
281
NEW 282	self.add_policies("p", filter.ptype, new_rules)	×
283
284	# return deleted rules
285
NEW 286	return old_rules	×

pycasbin / sqlalchemy-adapter / 2282901844

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous