How to use _relationships method in pandera

Best Python code snippet using pandera_python

wrapper.py

Source:wrapper.py

1import requests2from operator import attrgetter3import os4API_ROOT_URL = "https://api.tnyu.org/v3/"5class InvalidSearchAttributeError(Exception):6    """7    This exception is raised when in a sort method is invoked by an8    invalid attribute9    """10    def __init__(self, message):11        super(InvalidSearchAttributeError, self).__init__(message)12class AuthenticationException(Exception):13    """14    This exception is raised when a protected resource is accessed without15    authentication16    """17    pass18class TNYUAPI(object):19    def __init__(self, api_key=None):20        self.api_key = api_key21    def get_all_events(self, sort_by=None):22        """23        Return a list of Event objects for all events24        """25        resources = self.get_resource('events')['data']26        events = [Event.from_json(self, x) for x in resources]27        if sort_by:28            if not hasattr(events[0], sort_by):29                raise InvalidSearchAttributeError()30            return sorted(events, key=attrgetter(sort_by))31        return events32    def get_all_people(self, sort_by=None):33        resources = self.get_resource('people')['data']34        people = [Person.from_json(self, x) for x in resources]35        if sort_by:36            if not hasattr(people[0], sort_by):37                raise InvalidSearchAttributeError()38            return sorted(people, key=attrgetter(sort_by))39        return people40    def get_all_projects(self, sort_by=None):41        resources = self.get_resource('projects')['data']42        proj = [Project.from_json(self, x) for x in resources]43        if sort_by:44            if not hasattr(proj[0], sort_by):45                raise InvalidSearchAttributeError()46            return sorted(proj, key=attrgetter(sort_by))47        return proj48    def get_all_venues(self, sort_by=None):49        resources = self.get_resource('venues')['data']50        venues = [Venue.from_json(self, x) for x in resources]51        if sort_by:52            if not hasattr(venues[0], sort_by):53                raise InvalidSearchAttributeError()54            return sorted(venues, key=attrgetter(sort_by))55        return venues56    def get_all_organizations(self, sort_by=None):57        resources = self.get_resource('organizations')['data']58        org = [Organization.from_json(self, x) for x in resources]59        if sort_by:60            if not hasattr(org[0], sort_by):61                raise InvalidSearchAttributeError()62            return sorted(org, key=attrgetter(sort_by))63        return org64    def get_all_teams(self, sort_by=None):65        resources = self.get_resource('teams')['data']66        teams = [Team.from_json(self, x) for x in resources]67        if sort_by:68            if not hasattr(teams[0], sort_by):69                raise InvalidSearchAttributeError()70            return sorted(teams, key=attrgetter(sort_by))71        return teams72    def get_all_jobs(self, sort_by=None):73        resources = self.get_resource('jobs')['data']74        jobs = [Job.from_json(self, x) for x in resources]75        if sort_by:76            if not hasattr(jobs[0], sort_by):77                raise InvalidSearchAttributeError()78            return sorted(jobs, key=attrgetter(sort_by))79        return jobs80    def get_resource(self, path):81        headers = {82            'content-type': 'application/vnd.api+json',83            'accept': 'application/*, text/*',84            'authorization': 'Bearer ' + self.api_key85        }86        r = requests.get(API_ROOT_URL + path, headers=headers)87        return r.json()88class Job(object):89    def __init__(self, client, job_id, json_obj=None):90        self.id = job_id91        self.client = client92        if json_obj:93            self._attributes = json_obj['attributes']94            self._relationships = json_obj['relationships']95        else:96            # Pull data from the API using event id97            res = client.get_resource('jobs/%s' % self.id)['data']98            self._attributes = res['attributes']99            self._relationships = res['relationships']100    @classmethod101    def from_json(cls, client, json_obj):102        """103        Allow instantiation of an Job object from JSON104        """105        return Job(client, json_obj['id'], json_obj)106    def employer(self):107        employer_id = self._relationships['employer']['data']['id']108        return Organization(self.client, employer_id)109    def __getattr__(self, attr):110        if attr not in self._attributes:111            raise AttributeError("Job object has no attribute " + attr)112        return self._attributes[attr]113    def __repr__(self):114        return self.name115class Project(object):116    def __init__(self, client, proj_id, json_obj=None):117        self.id = proj_id118        self.client = client119        if json_obj:120            self._attributes = json_obj['attributes']121            self._relationships = json_obj['relationships']122        else:123            # Pull data from the API using event id124            res = client.get_resource('projects/%s' % self.id)['data']125            self._attributes = res['attributes']126            self._relationships = res['relationships']127    @classmethod128    def from_json(cls, client, json_obj):129        """130        Allow instantiation of an Project object from JSON131        """132        return Project(client, json_obj['id'], json_obj)133    def creators(self):134        creators_list = self._relationships['creators']['data']135        tmp = []136        for each_creator in creators_list:137            try:138                tmp.append(Person(self.client, each_creator['id']))139            except:140                continue141        return tmp142    def shown_at(self):143        events = self._relationships['shownAt']['data']144        return [Event(self.client, x['id']) for x in events]145    def __getattr__(self, attr):146        if attr not in self._attributes:147            raise AttributeError("Project object has no attribute " + attr)148        return self._attributes[attr]149class Organization(object):150    def __init__(self, client, org_id, json_obj=None):151        self.id = org_id152        self.client = client153        if json_obj:154            self._attributes = json_obj['attributes']155            self._relationships = json_obj['relationships']156        else:157            # Pull data from the API using event id158            res = client.get_resource('organizations/%s' % self.id)['data']159            self._attributes = res['attributes']160            self._relationships = res['relationships']161    @classmethod162    def from_json(cls, client, json_obj):163        """164        Allow instantiation of an Organization object from JSON165        """166        return Organization(client, json_obj['id'], json_obj)167    def liaisons(self):168        liaisons_list = self._relationships['liaisons']['data']169        tmp = []170        for each_person in liaisons_list:171            # Check if the person id exists in the API anymore172            try:173                tmp.append(Person(self.client, each_person['id']))174            except:175                continue176        return tmp177    def __getattr__(self, attr):178        if attr not in self._attributes:179            raise AttributeError("Org object has no attribute " + attr)180        return self._attributes[attr]181    def __repr__(self):182        return self.name183class Person(object):184    def __init__(self, client, person_id, json_obj=None):185        self.id = person_id186        self.client = client187        if json_obj:188            self._attributes = json_obj['attributes']189            self._relationships = json_obj['relationships']190        else:191            # Pull data from the API using event id192            res = client.get_resource('people/%s' % self.id)['data']193            self._attributes = res['attributes']194            self._relationships = res['relationships']195    @classmethod196    def from_json(cls, client, json_obj):197        """198        Allow instantiation of an Person object from JSON199        """200        return Person(client, json_obj['id'], json_obj)201    def organization(self):202        org_data = self._relationships['currentEmployer']['data']203        if org_data:204            return Organization(self.client, org_data['id'])205        return None206    def __getattr__(self, attr):207        if attr not in self._attributes:208            raise AttributeError("Person object has no attribute " + attr)209        return self._attributes[attr]210    def __repr__(self):211        return self.name212class Team(object):213    """214    This class represents a Tech @ NYU Team215    """216    def __init__(self, client, team_id, json_obj=None):217        self.id = team_id218        self.client = client219        if json_obj:220            self._attributes = json_obj['attributes']221        else:222            # Pull data from the API using event id223            res = client.get_resource('teams/%s' % self.id)['data']224            self._attributes = res['attributes']225    @classmethod226    def from_json(cls, client, json_obj):227        """228        Allow instantiation of an Team object from JSON229        """230        return Team(client, json_obj['id'], json_obj)231    def __getattr__(self, attr):232        if attr not in self._attributes:233            raise AttributeError("Team object has no attribute " + attr)234        return self._attributes[attr]235class Venue(object):236    """237    This class represents a Tech @ NYU event venue238    """239    def __init__(self, client, venue_id, json_obj=None):240        self.id = venue_id241        self.client = client242        if json_obj:243            self._attributes = json_obj['attributes']244            self._relationships = json_obj['relationships']245        else:246            # Pull data from the API using event id247            res = client.get_resource('venues/%s' % self.id)['data']248            self._attributes = res['attributes']249            self._relationships = res['relationships']250    @classmethod251    def from_json(cls, client, json_obj):252        """253        Allow instantiation of an Venue object from JSON254        """255        return Venue(client, json_obj['id'], json_obj)256    def __getattr__(self, attr):257        if attr not in self._attributes:258            raise AttributeError("Venue object has no attribute " + attr)259        return self._attributes[attr]260    def __repr__(self):261        return self.name262class Event(object):263    """264    Class to represent a Tech @ NYU Event265    """266    def __init__(self, client, event_id, json_obj=None):267        self.id = event_id268        self.client = client269        if json_obj:270            self._attributes = json_obj['attributes']271        else:272            # Pull data from the API using event id273            res = client.get_resource('events/%s' % self.id)['data']274            self._attributes = res['attributes']275            self._relationships = res['relationships']276    @classmethod277    def from_json(cls, client, json_obj):278        """279        Allow instantiation of an Event object from JSON280        """281        return Event(client, json_obj['id'], json_obj)282    def __getattr__(self, attr):283        if attr not in self._attributes:284            raise AttributeError("Event object has no attribute " + attr)285        return self._attributes[attr]286    def venue(self):287        venue_id = self._relationships['venue']['data']['id']288        return Venue(self.client, venue_id)289    def __repr__(self):290        return self.title.encode('utf-8')291if __name__ == '__main__':292    api = TNYUAPI(api_key=os.environ['TNYU_API_KEY'])...

relationship_store.py

Source:relationship_store.py

1from dataclasses import asdict, dataclass2from typing import Iterable, Iterator, Optional, TypedDict, cast3class RelationshipDict(TypedDict, total=False):4    list_item_id: str5    to_list_item_id: str6    relationship: str7@dataclass8class Relationship:9    """10    Represents a relationship between two items.11    """12    list_item_id: str13    to_list_item_id: str14    relationship: str15    def for_json(self) -> RelationshipDict:16        return cast(RelationshipDict, asdict(self))17class RelationshipStore:18    """19    Stores and updates relationships.20    """21    def __init__(22        self, relationships: Optional[Iterable[RelationshipDict]] = None23    ) -> None:24        self._is_dirty = False25        self._relationships = self._build_map(relationships or [])26    def __iter__(self) -> Iterator[Relationship]:27        return iter(self._relationships.values())28    def __contains__(self, relationship: Relationship) -> bool:29        return (30            relationship.list_item_id,31            relationship.to_list_item_id,32        ) in self._relationships33    def __len__(self) -> int:34        return len(self._relationships)35    @property36    def is_dirty(self) -> bool:37        return self._is_dirty38    def clear(self) -> None:39        self._relationships.clear()40        self._is_dirty = True41    def serialize(self) -> list[RelationshipDict]:42        return [43            relationship.for_json() for relationship in self._relationships.values()44        ]45    def get_relationship(46        self, list_item_id: str, to_list_item_id: str47    ) -> Optional[Relationship]:48        key = (list_item_id, to_list_item_id)49        return self._relationships.get(key)50    def remove_relationship(self, list_item_id: str, to_list_item_id: str) -> None:51        key = (list_item_id, to_list_item_id)52        if self._relationships.pop(key, None):53            self._is_dirty = True54    def add_or_update(self, relationship: Relationship) -> None:55        key = (relationship.list_item_id, relationship.to_list_item_id)56        existing_relationship = self._relationships.get(key)57        if existing_relationship != relationship:58            self._is_dirty = True59            self._relationships[key] = relationship60    def remove_all_relationships_for_list_item_id(self, list_item_id: str) -> None:61        """Remove all relationships associated with a particular list_item_id62        This method iterates through the entire list of relationships.63        """64        keys_to_delete = []65        for relationship in self:66            if list_item_id in (67                relationship.to_list_item_id,68                relationship.list_item_id,69            ):70                keys_to_delete.append(71                    (relationship.list_item_id, relationship.to_list_item_id)72                )73        for key in keys_to_delete:74            del self._relationships[key]75            self._is_dirty = True76    @staticmethod77    def _build_map(78        relationships: Iterable[RelationshipDict],79    ) -> dict[tuple[str, str], Relationship]:80        return {81            (82                relationship["list_item_id"],83                relationship["to_list_item_id"],84            ): Relationship(**relationship)85            for relationship in relationships...

event_relationship_model.py

Source:event_relationship_model.py

1class EventRelationship:2    def __init__(self):3        self._event_1 = None4        self._event_2 = None5        self._relationship_distribution = {}6        self._relationships = {}7    def to_dict(self):8        return {9            "event_1": self._event_1.to_dict(),10            "event_2": self._event_2.to_dict(),11            "relationship_distribution": self._relationship_distribution,12            "relationships": self._relationships,13        }14    def __repr__(self) -> str:15        return str(self.to_dict())16    def normalize_distribution(self):17        total = float(sum(self._relationships.values()))18        self._relationship_distribution = {k: v/total for k, v in self._relationships.items()}19    def relationship_distribution(self):20        return self._relationship_distribution21    def relationships(self):22        return self._relationships23    def add_relationship_type(self, relationship_type, relationship_score):24        self._relationships[relationship_type] = relationship_score25    def event_1(self):26        return self._event_127        28    def event_2(self):29        return self._event_230    @staticmethod31    def create(event_1, event_2):32        event_relationship = EventRelationship()33        event_relationship._event_1 = event_134        event_relationship._event_2 = event_2...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.