Company House API Integreation using python

import requests
import json
import datetime
import time
class CompaniesHouseService:
base_url = "https://api.companieshouse.gov.uk"
search_url = base_url+"/search/companies?q={}&items_per_page={}&start_index={}"
company_url = base_url+"/company/{}"
company_charges_url = base_url+"/company/{}/charges"
company_charge_url = base_url+"/company/{}/charges/{}"
company_exemptions_url = base_url+"/company/{}/exemptions"
company_filing_histories_url = base_url+"/company/{}/filing-history"
company_filing_history_url = base_url+"/company/{}/filing-history/{}"
company_insolvency_url = base_url+"/company/{}/insolvency"
person_corporate_entity_url = base_url+"/company/{}/persons-with-significant-control/corporate-entity/{}"
person_control_individual_url = base_url+"/company/{}/persons-with-significant-control/individual/{}"
person_control_legal_person_url = base_url+"/company/{}/persons-with-significant-control/legal-person/{}"
person_control_statement_url = base_url+"/company/{}/persons-with-significant-control-statements/{}"
person_control_super_secure_url = base_url+"/company/{}/persons-with-significant-control/super-secure/{}"
company_registers_url = base_url+"/company/{}/registers"
company_registered_office_address_url = base_url+"/company/{}/registered-office-address"
disqualified_officers_corporate_url = base_url+"/disqualified-officers/corporate/{}"
disqualified_officers_natural_url = base_url+"/disqualified-officers/natural/{}"
company_officers_url = base_url+"/company/{}/officers"
company_officer_url = base_url+"/company/{}/appointments/{}"
company_uk_establishments_url = base_url+"/company/{}/uk-establishments"
company_officer_appointments_url = base_url+"/officers/{}/appointments?items_per_page={}&start_index={}"
person_significate_control_url = base_url+"/company/{}/persons-with-significant-control?items_per_page={}&start_index={}"
person_signicate_control_statements_url = base_url+"/company/{}/persons-with-significant-control-statements?items_per_page={}&start_index={}"
def __init__(self, key, time_between_requests=0.5):
self.key = key
self.time_between_requests = time_between_requests
self.last_request_timestamp = None
def query_companieshouse_api(self, url, query, items_per_page=10, start_index=0):
query = self._remove_problem_characters(query)
self._rate_limiting()
resultQuery = requests.get(url.format(query, items_per_page, start_index),auth=(self.key,''))
print(url.format(query, items_per_page, start_index))
if resultQuery.status_code == 200:
result = json.JSONDecoder().decode(resultQuery.text)
else:
print(f"Failed with error code: {resultQuery.status_code} | "\
f"Reason: {resultQuery.reason}")
result = {}
return result
def _rate_limiting(self):
if self.last_request_timestamp is None:
self.last_request_timestamp = datetime.datetime.now()
else:
current_time = datetime.datetime.now()
time_since_request = (current_time - self.last_request_timestamp).total_seconds()
wait_time = max(self.time_between_requests - time_since_request, 0)
time.sleep(wait_time)
self.last_request_timestamp = datetime.datetime.now()
def _remove_problem_characters(self, string):
string = string.replace(" ","+")
string = string.replace("&","%26")
return string
def get_company_profile(self, company_number):
company_profile = self.query_companieshouse_api(self.company_url, company_number)
return company_profile
def get_company_charges(self, company_number, charge_id):
company_charges = self.query_companieshouse_api(self.company_charge_url, company_number, charge_id)
return company_charges
def list_company_charges(self, company_number):
company_charges_list = self.query_companieshouse_api(self.company_charges_url, company_number)
return company_charges_list
def get_company_exemptions(self, company_number):
company_exemptions = self.query_companieshouse_api(self.company_exemptions_url, company_number)
return company_exemptions
def list_company_filing_history(self, company_number):
company_filing_histories = self.query_companieshouse_api(self.company_filing_histories_url, company_number)
return company_filing_histories
def get_company_filing_history(self, company_number, transaction_id):
company_filing_history = self.query_companieshouse_api(self.company_filing_history_url, company_number, transaction_id)
return company_filing_history
def get_company_insolvency(self, company_number):
company_insolvency = self.query_companieshouse_api(self.company_insolvency_url, company_number)
return company_insolvency
def get_company_persons_with_significant_control_corporate_entity(self, company_number,psc_id):
person_corporate_entity = self.query_companieshouse_api(self.person_corporate_entity_url, company_number)
return person_corporate_entity
def get_company_persons_with_significant_control_individual(self, company_number, psc_id):
person_control_individual = self.query_companieshouse_api(self.person_control_individual_url, company_number)
return person_control_individual
def get_company_persons_with_significant_control_legal_person(self, company_number, psc_id):
person_control_legal_person = self.query_companieshouse_api(self.person_control_legal_person_url, company_number)
return person_control_legal_person
def get_company_persons_with_significant_control_statements(self, company_number, statement_id):
person_control_statement = self.query_companieshouse_api(self.person_control_statement_url, company_number)
return person_control_statement
def get_company_persons_with_significant_control_super_secure(self, company_number, super_secure_id):
person_control_super_secure = self.query_companieshouse_api(self.person_control_super_secure_url, company_number)
return person_control_super_secure
def get_company_registered_office_address(self, company_number):
company_registered_office_address = self.query_companieshouse_api(self.company_registered_office_address_url, company_number)
return company_registered_office_address
def get_company_registers(self, company_number):
company_registers = self.query_companieshouse_api(self.company_registers_url, company_number)
return company_registers
def get_disqualified_officers_corporate(self, officer_id):
disqualified_officers_corporate = self.query_companieshouse_api(self.disqualified_officers_corporate_url, company_number)
return disqualified_officers_corporate
def get_disqualified_officers_natural(self, officer_id):
disqualified_officers_natural = self.query_companieshouse_api(self.disqualified_officers_natural_url, company_number)
return disqualified_officers_natural
def list_company_officers(self, company_number):
company_officers = self.query_companieshouse_api(self.company_officers_url, company_number)
return company_officers
def get_company_officers(self, company_number, appointment_id):
company_officer = self.query_companieshouse_api(self.company_officer_url, company_number)
return company_officer
def list_company_uk_establishments(self, company_number):
company_uk_establishments = self.query_companieshouse_api(self.company_uk_establishments_url, company_number)
return company_uk_establishments
def list_company_persons_with_significant_control(self, company_number, items_per_page=10, start_index=0):
person_significate_control = self.query_companieshouse_api(self.person_significate_control_url, company_number, items_per_page, start_index)
return person_significate_control
def list_company_persons_with_significant_control_statements(self, company_number, items_per_page=10, start_index=0):
person_signicate_control_statements = self.query_companieshouse_api(self.person_signicate_control_statements_url, company_number)
return person_signicate_control_statements
def list_officers_appointments(self, officer_id, items_per_page=10, start_index=0):
company_officer_appointments = self.query_companieshouse_api(self.company_officer_appointments_url, company_number, items_per_page, start_index)
return company_officer_appointments
def search_companies(self, company_name, items_per_page=10, start_index=0):
search_result = self.query_companieshouse_api(self.search_url, company_name, items_per_page, start_index)
return search_result
if __name__ == "__main__":
key = "XXXXXXXXXXXXXXXXXXXXX" #Generate key https://developer.company-information.service.gov.uk/
ch_api = CompaniesHouseService(key)
search_query = "google"
company_number = "00445790"
charge_id = "5QU1lSudRI2jTIfUv_AOVxfLxVE"
transaction_id = "MzMyODc4MjQ4MmFkaXF6a2N4"
company_profile = ch_api.get_company_profile(company_number)
companies = ch_api.search_companies(search_query, 100, 0)
company_charges_list = ch_api.list_company_charges(company_number)
company_charges = ch_api.get_company_charges(company_number, charge_id)
company_exemptions = ch_api.get_company_exemptions(company_number)
company_filing_histories = ch_api.list_company_filing_history(company_number)
company_filing_history = ch_api.get_company_filing_history(company_number, transaction_id)
company_insolvency = ch_api.get_company_insolvency(company_number)

Cloning a repository using the command line

$ git clone { repository name(https://github.com/username/Projectname) }

Repository name--> Copy from github.com-->Code-->Https

Error while cloning--- 'git' is not recognized as an internal or external command, operable program

Already installed in Linux--> No Need to do any process

Windows-->

1- Install Git -- https://git-scm.com/download/win

2-Set Path

Go to Start

Start typing 'This PC'

Right-click This PC, choose Properties

On the left side of the window that pops up, click on Advanced System Settings

Click on the Advanced tab

Click on the Environmental Variables button at the bottom

Down in the System Variables section, double-click Path

Click the New button in the top right corner

Add this path: C:\Program Files\Git\bin\ then click the enter key

Add another path: C:\Program Files\Git\cmd

Close & re-open the console if it's already open.

Locust Module: Python Script fir Locust performance testing tool

from locust import HttpUser, task, between

access_token = "XXXXXX"

class QuickstartUser(HttpUser):

wait_time = between(1, 2)

def on_start(self):

payload = { "access_token":access_token, "token_type": "Bearer"}

response = self.client.post("endpoint1", json=payload)

print(response.status_code)

print(response.__dict__)

print("==============================================================")

@task

def get_account(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint2", headers=headers)

print("==============================================================")

@task

def get_all_account(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint3", headers=headers)

print("==============================================================")

@task

def get_user_features(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint4", headers=headers)

print("==============================================================")

@task

def get_app_version(self):

response = self.client.get("/endpoint5")

print("==============================================================")

@task

def get_case_status(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint6", headers=headers)

print("==============================================================")

@task

def fun7(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint7", headers=headers)

print("jeet.........", response.__dict__)

print("==============================================================")

@task

def fun8(self):

headers = {"Content-type": "application/json", "Authorization": "Bearer " + access_token}

response = self.client.get("/endpoint8", headers=headers)

print("==============================================================")

# @task(3)

# def view_item(self):

# for item_id in range(10):

# self.client.get(f"/item?id={item_id}", name="/item")

PUT Method Using FastAPI - JSON Server

@app.put("/user/update/{user_id}")

def update_user(user_id: str, user: UserCreate):

with open("test.json", 'r') as file:

dict = json.load(file)

if user_id in dict:

update_item_encoded = jsonable_encoder(user)

dict[user_id] = update_item_encoded

data = json.dumps(dict, indent=4)

print(data)

with open("test.json", 'w') as file:

file.write(data)

return "Updated Record"