Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use get_project method in dbt-osmosis

Best Python code snippet using dbt-osmosis_python

project_getter.py

Source:project_getter.py

...63            "HAS_COMMENTS", self.ld["get_project"]["hasCommentsReplacer"] if details["hasComments"] else self.ld["get_project"]["noCommentsReplacer"]64        )65        print(output)66        return output, details["thumbnails"]  # returns thumbnails too for later use67    def get_project(self, project_id):68        # Connecting to the API69        log(self.ld["reusable"]["connectingToAPIAwaiter"])70        response = requests.get(API_PROJECT.replace("~", project_id))71        insert_success(self.ld)72        # Loading data73        log(self.ld["reusable"]["collectingDataAwaiter"])74        json_data = response.json()75        insert_success(self.ld)76        # Checking if the user exists77        log("Validating project existence...")78        try:79            json_data["id"]  # 404 responses do not have IDs80        except KeyError:81            return None82        # Creating dictionary, loading title83        project_data = {"project_id": json_data["id"]}84        log(self.ld["get_project"]["loadingTitleAwaiter"])85        project_data["title"] = json_data["title"]86        insert_success(self.ld)87        # Loading visibility88        log(self.ld["get_project"]["loadingVisibilityAwaiter"])89        project_data["visibility"] = json_data["is_published"]  # check to see the other options90        insert_success(self.ld)91        # Determining if the project is a remix92        log(self.ld["get_project"]["loadingIsRemixAwaiter"])93        if json_data["remix"]["parent"] is not None:94            project_data["isRemix"] = True95            project_data["parentID"] = json_data["remix"]["parent"]96            project_data["rootID"] = json_data["remix"]["root"]97        else:98            project_data["isRemix"] = False99        insert_success(self.ld)100        # Loading author101        log(self.ld["get_project"]["loadingAuthorAwaiter"])102        project_data["author"] = json_data["author"]["username"]103        insert_success(self.ld)104        # Loading token105        log(self.ld["get_project"]["loadingTokenAwaiter"])106        project_data["token"] = json_data["project_token"]107        insert_success(self.ld)108        # Loading dates109        log(self.ld["get_project"]["loadingCreationDateAwaiter"])110        project_data["creationDate"] = self.format_date(json_data["history"]["created"], self.ld)111        insert_success(self.ld)112        log(self.ld["get_project"]["loadingEditDateAwaiter"])113        project_data["lastEditedDate"] = self.format_date(json_data["history"]["modified"], self.ld)114        insert_success(self.ld)115        log(self.ld["get_project"]["loadingPublishDateAwaiter"])116        project_data["lastPublishedDate"] = self.format_date(json_data["history"]["shared"],117                                                             self.ld)  # Some unpublished projects may not have valid info118        insert_success(self.ld)119        # Loading stats120        log(self.ld["get_project"]["loadingViewsAwaiter"])121        project_data["views"] = json_data["stats"]["views"]122        insert_success(self.ld)123        log(self.ld["get_project"]["loadingLovesAwaiter"])124        project_data["loves"] = json_data["stats"]["loves"]125        insert_success(self.ld)126        log(self.ld["get_project"]["loadingFavouritesAwaiter"])127        project_data["faves"] = json_data["stats"]["favorites"]  # UK English > US English no cap128        insert_success(self.ld)129        log(self.ld["get_project"]["loadingRemixesAwaiter"])130        project_data["remixes"] = json_data["stats"]["remixes"]131        insert_success(self.ld)132        # Loading instructions and description (lengthy parts)133        log(self.ld["get_project"]["loadingInstructionsAwaiter"])134        project_data["instructions"] = json_data["instructions"]135        insert_success(self.ld)136        log(self.ld["get_project"]["loadingDescriptionAwaiter"])137        project_data["description"] = json_data["description"]138        insert_success(self.ld)139        # Finally140        log(self.ld["get_project"]["loadingHasCommentsAwaiter"])141        project_data["hasComments"] = json_data["comments_allowed"]142        insert_success(self.ld)143        log(self.ld["get_project"]["loadingThumbnailsAwaiter"])144        project_data["thumbnails"] = {"480x360": json_data["image"], **json_data["images"]}145        insert_success(self.ld)146        # Returning all data147        return project_data148    def begin_project_process(self):149        # Getting user input150        project_id = questionary.text(self.ld["get_project"]["enterProjectIDQuery"],151                                      validate=lambda x: x.isnumeric()).unsafe_ask().strip()152        # Getting project content153        clear()154        print(self.ld["get_project"]["projectID"], project_id)155        project_data = self.get_project(project_id)156        # Checking if the project actually exists157        if project_data is None:158            insert_error(self.ld)159            print(self.ld["get_project"]["projectInexistent"].replace("PROJECT_ID", project_id))160            await_enter(self.ld, to_exit=True)161            clear()162            return163        # Showing all content164        clear()165        formatted_response, thumbnails = self.reveal_project(project_data)166        # File saving option167        print(self.ld["get_project"]["thumbnailAvailability"])168        should_save = questionary.select(169            self.ld["reusable"]["saveFileQuery"],...

unit_test.py

Source:unit_test.py

...22@pytest.fixture()23def project_manager(mocker, authenticated_user):24    mocker.patch('portal.apps.projects.managers.base.ProjectsManager.get_project')25    project = ProjectsManager(authenticated_user)26    project.get_project().project_id = "PRJ-123"27    project.get_project().storage.storage.root_dir = os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123")28    return project29def test_search(mocker, authenticated_user, project_manager, mock_index):30    mock_listing = mocker.patch('portal.apps.projects.managers.base.ProjectsManager.list')31    mock_listing.return_value = []32    mock_index.search().query().execute().return_value = []33    project_manager.search('testquery')34    assert mock_listing.call_count == 135    mock_index.search().query.assert_called_with('query_string',36                                                 query='testquery',37                                                 minimum_should_match="80%")38def test_add_member_pi(authenticated_user, project_manager, service_account):39    """Test add a PI to a project."""40    project_manager.add_member('PRJ-123', 'pi', 'username')41    project_manager.get_project().add_member.assert_not_called()42    project_manager.get_project().add_co_pi.assert_not_called()43    project_manager.get_project().add_pi.assert_called_with(authenticated_user)44    service_account().jobs.submit.assert_called_with(45        body={46            "name": "username-PRJ-123-acls",47            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,48            "archive": False,49            "parameters": {50                "projectId": "PRJ-123",51                "username": "username",52                "action": "add",53                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),54            }55        }56    )57def test_add_member_co_pi(authenticated_user, project_manager, service_account):58    """Test add a PI to a project."""59    project_manager.add_member('PRJ-123', 'co_pi', 'username')60    project_manager.get_project().add_member.assert_not_called()61    project_manager.get_project().add_pi.assert_not_called()62    project_manager.get_project().add_co_pi.assert_called_with(authenticated_user)63    service_account().jobs.submit.assert_called_with(64        body={65            "name": "username-PRJ-123-acls",66            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,67            "archive": False,68            "parameters": {69                "projectId": "PRJ-123",70                "username": "username",71                "action": "add",72                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),73            }74        }75    )76def test_add_member(authenticated_user, project_manager, service_account):77    """Test add a PI to a project."""78    project_manager.add_member('PRJ-123', 'team_member', 'username')79    project_manager.get_project().add_co_pi.assert_not_called()80    project_manager.get_project().add_pi.assert_not_called()81    project_manager.get_project().add_member.assert_called_with(authenticated_user)82    service_account().jobs.submit.assert_called_with(83        body={84            "name": "username-PRJ-123-acls",85            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,86            "archive": False,87            "parameters": {88                "projectId": "PRJ-123",89                "username": "username",90                "action": "add",91                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),92            }93        }94    )95def test_remove_member_pi(authenticated_user, project_manager, service_account):96    """Test add a PI to a project."""97    project_manager.remove_member('PRJ-123', 'pi', 'username')98    project_manager.get_project().remove_member.assert_not_called()99    project_manager.get_project().remove_co_pi.assert_not_called()100    project_manager.get_project().remove_pi.assert_called_with(authenticated_user)101    service_account().jobs.submit.assert_called_with(102        body={103            "name": "username-PRJ-123-acls",104            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,105            "archive": False,106            "parameters": {107                "projectId": "PRJ-123",108                "username": "username",109                "action": "remove",110                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),111            }112        }113    )114def test_remove_member_co_pi(authenticated_user, project_manager, service_account):115    """Test add a PI to a project."""116    project_manager.remove_member('PRJ-123', 'co_pi', 'username')117    project_manager.get_project().remove_member.assert_not_called()118    project_manager.get_project().remove_pi.assert_not_called()119    project_manager.get_project().remove_co_pi.assert_called_with(authenticated_user)120    service_account().jobs.submit.assert_called_with(121        body={122            "name": "username-PRJ-123-acls",123            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,124            "archive": False,125            "parameters": {126                "projectId": "PRJ-123",127                "username": "username",128                "action": "remove",129                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),130            }131        }132    )133def test_remove_member(authenticated_user, project_manager, service_account):134    """Test add a PI to a project."""135    project_manager.remove_member('PRJ-123', 'team_member', 'username')136    project_manager.get_project().remove_co_pi.assert_not_called()137    project_manager.get_project().remove_pi.assert_not_called()138    project_manager.get_project().remove_member.assert_called_with(authenticated_user)139    service_account().jobs.submit.assert_called_with(140        body={141            "name": "username-PRJ-123-acls",142            "appId": settings.PORTAL_PROJECTS_PEMS_APP_ID,143            "archive": False,144            "parameters": {145                "projectId": "PRJ-123",146                "username": "username",147                "action": "remove",148                "root_dir": os.path.join(settings.PORTAL_PROJECTS_ROOT_DIR, "PRJ-123"),149            }150        }151    )152def test_project_manager_create(mocker, authenticated_user, project_manager, portal_project, mock_project_save_signal):...

__main__.py

Source:__main__.py

2import pulumi_hcp as hcp3import pulumi_aws as aws4def main() -> None:5    aws_tags = {6        "pulumi:project": pulumi.get_project(),7        "pulumi:stack": pulumi.get_stack()8    }9    # Create the HVN (Hashicorp Virtual Network)10    hvn = hcp.Hvn(11        f"{pulumi.get_project()}-hvn",12        hvn_id=f"{pulumi.get_project()}-hvn",13        cloud_provider="aws",14        region="us-east-1",15        cidr_block="172.25.16.0/20",16    )17    # Create an AWS VPC to peer with the HVN18    peer_vpc = aws.ec2.Vpc(19        f"{pulumi.get_project()}-vpc",20        cidr_block="172.31.0.0/16",21        tags={22            "Name": f"{pulumi.get_project()}-vpc",23            **aws_tags,24        },25    )26    # Initiate the peering connection from HCP27    peering_connection = hcp.AwsNetworkPeering(28        f"{pulumi.get_project()}-peering",29        hvn_id=hvn.hvn_id,30        peering_id=f"{pulumi.get_project()}-peering",31        peer_vpc_id=peer_vpc.id,32        peer_account_id=peer_vpc.owner_id,33        peer_vpc_region="us-east-1",34        opts=pulumi.ResourceOptions(custom_timeouts=pulumi.CustomTimeouts(create='10m'))35    )36    # Accept the peering connection in AWS37    connection_accepter = aws.ec2.VpcPeeringConnectionAccepter(38        f"{pulumi.get_project()}-connection-accepter",39        vpc_peering_connection_id=peering_connection.provider_peering_id,40        auto_accept=True,41        tags=aws_tags,42    )43    # Create a route in the HVN to resources in the VPC44    hcp.HvnRoute(45        f"{pulumi.get_project()}-route-to-vpc",46        hvn_link=hvn.self_link,47        hvn_route_id=f"{pulumi.get_project()}-route-to-vpc",48        destination_cidr=peer_vpc.cidr_block,49        target_link=peering_connection.self_link50    )51    # Create a route in one of the VPCs routing tables (here, we'll52    # just use the main route table, but any routing table is fair53    # game) to the HVN.54    aws.ec2.Route(55        f"{pulumi.get_project()}-route-to-hvn",56        route_table_id=peer_vpc.main_route_table_id,57        destination_cidr_block=hvn.cidr_block,58        # Depend on the accepter, for proper sequencing59        vpc_peering_connection_id=connection_accepter.vpc_peering_connection_id,60    )61if __name__ == "__main__":...

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.