6415123940

Committed 05 Oct 2023 05:45AM UTC coverage: 79.915%. First build

Build # 6415123940

Build Type

Pull #70

github

Committed by

web-flow

Commit Message

Merge 41f53390c into f034f9a27

Pull Request Pull Request #70: build: Updated pydantic version to 2.4

Run Details

117 of 117 new or added lines in 19 files covered. (100.0%)

1886 of 2360 relevant lines covered (79.92%)

0.8 hits per line

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

81.48

/src/unipoll_api/documents.py

# from typing import ForwardRef, NewType, TypeAlias, Optional
from typing import Literal
from bson import DBRef
from beanie import BackLink, Document, WriteRules, after_event, Insert, Link, PydanticObjectId  # BackLink
from fastapi_users_db_beanie import BeanieBaseUser
from pydantic import Field
from unipoll_api.utils import colored_dbg as Debug
from unipoll_api.utils.token_db import BeanieBaseAccessToken


# Create a link to the Document model
async def create_link(document: Document) -> Link:
    ref = DBRef(collection=document._document_settings.name,  # type: ignore
                id=document.id)
    link = Link(ref, type(document))
    return link


# Custom PydanticObjectId class to override due to a bug
class ResourceID(PydanticObjectId):
    @classmethod
    def __modify_schema__(cls, field_schema):  # type: ignore
        field_schema.update(
            type="string",
            example="5eb7cf5a86d9755df3a6c593",
        )


class AccessToken(BeanieBaseAccessToken, Document):  # type: ignore
    pass


class Resource(Document):
    id: ResourceID = Field(default_factory=ResourceID, alias="_id")
    resource_type: Literal["workspace", "group", "poll"]
    name: str = Field(title="Name", description="Name of the resource", min_length=3, max_length=50)
    description: str = Field(default="", title="Description", max_length=1000)
    policies: list[Link["Policy"]] = []

    @after_event(Insert)
    def create_group(self) -> None:
        Debug.info(f'New {self.resource_type} "{self.id}" has been created')

    async def add_member(self, account: "Account", permissions, save: bool = True) -> "Account":
        # Add the account to the group
        self.members.append(account)  # type: ignore
        # Create a policy for the new member
        new_policy = Policy(policy_holder_type='account',
                            policy_holder=(await create_link(account)),
                            permissions=permissions,
                            parent_resource=self)  # type: ignore

        # Add the policy to the group
        self.policies.append(new_policy)  # type: ignore
        if save:
            await Resource.save(self, link_rule=WriteRules.WRITE)
        return account

    async def remove_member(self, account, save: bool = True) -> bool:
        # Remove the account from the group
        # await self.fetch_link("members")
        for i, member in enumerate(self.members):
            if account.id == member.id:  # type: ignore
                self.members.remove(member)
                Debug.info(f"Removed member {member.id} from {self.resource_type} {self.id}")
                break

        # Remove the policy from the group
        # await self.fetch_link("policies")
        for policy in self.policies:
            # pc = await policy.policy_holder.fetch()  # type: ignore
            pc = policy.policy_holder  # type: ignore
            if pc.ref.id == account.id:
                self.policies.remove(policy)
                await Policy.delete(policy)
                Debug.info(f"Removed policy: {pc.ref.id} from {self.resource_type} {self.id}")
                break

        await self.save(link_rule=WriteRules.WRITE)  # type: ignore
        return True


class Account(BeanieBaseUser, Document):  # type: ignore
    id: ResourceID = Field(default_factory=ResourceID, alias="_id")
    first_name: str = Field(
        default_factory=str,
        max_length=20,
        min_length=2,
        pattern="^[A-Z][a-z]*$")
    last_name: str = Field(
        default_factory=str,
        max_length=20,
        min_length=2,
        pattern="^[A-Z][a-z]*$")


class Workspace(Resource):
    resource_type: Literal["workspace"] = "workspace"
    members: list[Link["Account"]] = []
    groups: list[Link["Group"]] = []
    polls: list[Link["Poll"]] = []


class Group(Resource):
    resource_type: Literal["group"] = "group"
    workspace: BackLink[Workspace] = Field(original_field="groups")
    members: list[Link["Account"]] = []
    groups: list[Link["Group"]] = []


class Policy(Document):
    id: ResourceID = Field(default_factory=ResourceID, alias="_id")
    parent_resource: BackLink["Workspace"] | BackLink["Group"] | BackLink["Poll"] = Field(original_field="policies")
    policy_holder_type: Literal["account", "group"]
    policy_holder: Link["Group"] | Link["Account"]
    permissions: int


class Poll(Resource):
    id: ResourceID = Field(default_factory=ResourceID, alias="_id")
    workspace: BackLink["Workspace"] = Field(original_field="polls")
    resource_type: Literal["poll"] = "poll"
    public: bool
    published: bool
    questions: list
    policies: list[Link["Policy"]]


# NOTE: model_rebuild is used to avoid circular imports
Resource.model_rebuild()
Workspace.model_rebuild()
Group.model_rebuild()
Policy.model_rebuild()
Poll.model_rebuild()

1	# from typing import ForwardRef, NewType, TypeAlias, Optional
2	from typing import Literal	1✔
3	from bson import DBRef	1✔
4	from beanie import BackLink, Document, WriteRules, after_event, Insert, Link, PydanticObjectId # BackLink	1✔
5	from fastapi_users_db_beanie import BeanieBaseUser	1✔
6	from pydantic import Field	1✔
7	from unipoll_api.utils import colored_dbg as Debug	1✔
8	from unipoll_api.utils.token_db import BeanieBaseAccessToken	1✔
9
10
11	# Create a link to the Document model
12	async def create_link(document: Document) -> Link:	1✔
13	ref = DBRef(collection=document._document_settings.name, # type: ignore	1✔
14	id=document.id)
15	link = Link(ref, type(document))	1✔
16	return link	1✔
17
18
19	# Custom PydanticObjectId class to override due to a bug
20	class ResourceID(PydanticObjectId):	1✔
21	@classmethod	1✔
22	def __modify_schema__(cls, field_schema): # type: ignore	1✔
23	field_schema.update(	×
24	type="string",
25	example="5eb7cf5a86d9755df3a6c593",
26	)
27
28
29	class AccessToken(BeanieBaseAccessToken, Document): # type: ignore	1✔
30	pass	1✔
31
32
33	class Resource(Document):	1✔
34	id: ResourceID = Field(default_factory=ResourceID, alias="_id")	1✔
35	resource_type: Literal["workspace", "group", "poll"]	1✔
36	name: str = Field(title="Name", description="Name of the resource", min_length=3, max_length=50)	1✔
37	description: str = Field(default="", title="Description", max_length=1000)	1✔
38	policies: list[Link["Policy"]] = []	1✔
39
40	@after_event(Insert)	1✔
41	def create_group(self) -> None:	1✔
42	Debug.info(f'New {self.resource_type} "{self.id}" has been created')	1✔
43
44	async def add_member(self, account: "Account", permissions, save: bool = True) -> "Account":	1✔
45	# Add the account to the group
46	self.members.append(account) # type: ignore	1✔
47	# Create a policy for the new member
48	new_policy = Policy(policy_holder_type='account',	1✔
49	policy_holder=(await create_link(account)),
50	permissions=permissions,
51	parent_resource=self) # type: ignore
52
53	# Add the policy to the group
54	self.policies.append(new_policy) # type: ignore	1✔
55	if save:	1✔
56	await Resource.save(self, link_rule=WriteRules.WRITE)	1✔
57	return account	1✔
58
59	async def remove_member(self, account, save: bool = True) -> bool:	1✔
60	# Remove the account from the group
61	# await self.fetch_link("members")
62	for i, member in enumerate(self.members):	×
63	if account.id == member.id: # type: ignore	×
64	self.members.remove(member)	×
65	Debug.info(f"Removed member {member.id} from {self.resource_type} {self.id}")	×
66	break	×
67
68	# Remove the policy from the group
69	# await self.fetch_link("policies")
70	for policy in self.policies:	×
71	# pc = await policy.policy_holder.fetch() # type: ignore
72	pc = policy.policy_holder # type: ignore	×
73	if pc.ref.id == account.id:	×
74	self.policies.remove(policy)	×
75	await Policy.delete(policy)	×
76	Debug.info(f"Removed policy: {pc.ref.id} from {self.resource_type} {self.id}")	×
77	break	×
78
79	await self.save(link_rule=WriteRules.WRITE) # type: ignore	×
80	return True	×
81
82
83	class Account(BeanieBaseUser, Document): # type: ignore	1✔
84	id: ResourceID = Field(default_factory=ResourceID, alias="_id")	1✔
85	first_name: str = Field(	1✔
86	default_factory=str,
87	max_length=20,
88	min_length=2,
89	pattern="^[A-Z][a-z]*$")
90	last_name: str = Field(	1✔
91	default_factory=str,
92	max_length=20,
93	min_length=2,
94	pattern="^[A-Z][a-z]*$")
95
96
97	class Workspace(Resource):	1✔
98	resource_type: Literal["workspace"] = "workspace"	1✔
99	members: list[Link["Account"]] = []	1✔
100	groups: list[Link["Group"]] = []	1✔
101	polls: list[Link["Poll"]] = []	1✔
102
103
104	class Group(Resource):	1✔
105	resource_type: Literal["group"] = "group"	1✔
106	workspace: BackLink[Workspace] = Field(original_field="groups")	1✔
107	members: list[Link["Account"]] = []	1✔
108	groups: list[Link["Group"]] = []	1✔
109
110
111	class Policy(Document):	1✔
112	id: ResourceID = Field(default_factory=ResourceID, alias="_id")	1✔
113	parent_resource: BackLink["Workspace"] \| BackLink["Group"] \| BackLink["Poll"] = Field(original_field="policies")	1✔
114	policy_holder_type: Literal["account", "group"]	1✔
115	policy_holder: Link["Group"] \| Link["Account"]	1✔
116	permissions: int	1✔
117
118
119	class Poll(Resource):	1✔
120	id: ResourceID = Field(default_factory=ResourceID, alias="_id")	1✔
121	workspace: BackLink["Workspace"] = Field(original_field="polls")	1✔
122	resource_type: Literal["poll"] = "poll"	1✔
123	public: bool	1✔
124	published: bool	1✔
125	questions: list	1✔
126	policies: list[Link["Policy"]]	1✔
127
128
129	# NOTE: model_rebuild is used to avoid circular imports
130	Resource.model_rebuild()	1✔
131	Workspace.model_rebuild()	1✔
132	Group.model_rebuild()	1✔
133	Policy.model_rebuild()	1✔
134	Poll.model_rebuild()	1✔

unipoll / api / 6415123940

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