"""Arcadium Data API client for Python (API v1.39.1). Generated from the OpenAPI contract by tools/src/gen-clients.ts. Do not edit by hand. Python 3.9+, standard library only. from arcadium_client import ArcadiumClient api = ArcadiumClient(api_key=os.environ.get("ARCADIUM_API_KEY")) # api_key is optional r = api.get_company_pulse("stripe.com") print(r.data, r.rate_limit.remaining) """ import json import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from typing import Any, Callable, Optional from urllib.parse import quote API_VERSION = "1.39.1" DEFAULT_BASE_URL = "https://api.arcadiumdata.io" def _num(v: Optional[str]) -> Optional[float]: try: return float(v) if v not in (None, "") else None except ValueError: return None @dataclass class RateLimit: """Rate-limit headers of a response (None when absent).""" remaining: Optional[float] = None reset: Optional[float] = None quota_limit: Optional[float] = None quota_remaining: Optional[float] = None retry_after: Optional[float] = None anonymous: bool = False @dataclass class ApiResponse: status: int data: Any rate_limit: RateLimit class ArcadiumError(Exception): """Every non-2xx answer; `code` is the API's `error` value (see /docs/errors/).""" def __init__(self, status: int, code: str, message: str, rate_limit: RateLimit): super().__init__(f"{status} {code}: {message}") self.status = status self.code = code self.message = message self.rate_limit = rate_limit @property def retryable(self) -> bool: """Worth retrying after `rate_limit.retry_after` seconds (429, 500, 503).""" return self.status in (429, 500, 503) def _rate_limit(h: Any) -> RateLimit: return RateLimit( remaining=_num(h.get("x-ratelimit-remaining")), reset=_num(h.get("x-ratelimit-reset")), quota_limit=_num(h.get("x-quota-limit")), quota_remaining=_num(h.get("x-quota-remaining")), retry_after=_num(h.get("retry-after")), anonymous=h.get("x-arcadium-anon") == "true", ) class ArcadiumClient: def __init__( self, api_key: Optional[str] = None, base_url: str = DEFAULT_BASE_URL, timeout: float = 30.0, opener: Optional[Callable[..., Any]] = None, ): self.api_key = api_key self.base_url = base_url.rstrip("/") self.timeout = timeout self._open = opener or urllib.request.urlopen def _request(self, method: str, path: str, params: Optional[dict] = None, body: Any = None) -> ApiResponse: query = urllib.parse.urlencode( { k: (str(v).lower() if isinstance(v, bool) else v) for k, v in (params or {}).items() if v is not None and v != "" } ) url = self.base_url + path + ("?" + query if query else "") headers = {"Accept": "application/json"} if self.api_key: headers["Authorization"] = "Bearer " + self.api_key data = None if body is not None: headers["Content-Type"] = "application/json" data = json.dumps(body).encode("utf-8") req = urllib.request.Request(url, data=data, headers=headers, method=method) try: res = self._open(req, timeout=self.timeout) status, raw, hdrs = res.status, res.read(), res.headers except urllib.error.HTTPError as e: status, raw, hdrs = e.code, e.read(), e.headers text = raw.decode("utf-8") if raw else "" try: parsed = json.loads(text) if text else None except ValueError: parsed = text rl = _rate_limit({k.lower(): v for k, v in hdrs.items()}) if status >= 400: err = parsed if isinstance(parsed, dict) else {} raise ArcadiumError(status, err.get("error", "unavailable"), err.get("message", ""), rl) return ApiResponse(status, parsed, rl) def search_jobs(self, q=None, country=None, ats=None, remote=None, domain=None, posted_since=None, seen_since=None, isco=None, limit=None, cursor=None): """Search open jobs (GET /v1/jobs)""" return self._request("GET", "/v1/jobs", {"q": q, "country": country, "ats": ats, "remote": remote, "domain": domain, "postedSince": posted_since, "seenSince": seen_since, "isco": isco, "limit": limit, "cursor": cursor}) def search_remote_jobs(self, q=None, region=None, days=None, limit=None, cursor=None): """Search open remote jobs, direct from employers (GET /v1/jobs/remote)""" return self._request("GET", "/v1/jobs/remote", {"q": q, "region": region, "days": days, "limit": limit, "cursor": cursor}) def search_recalls(self, q=None, source=None, country=None, category=None, brand=None, risk_type=None, barcode=None, since=None, until=None, with_matches=None, limit=None, cursor=None): """Recall Radar: product safety alerts (EU Safety Gate, US CPSC, UK OPSS) (GET /v1/recalls)""" return self._request("GET", "/v1/recalls", {"q": q, "source": source, "country": country, "category": category, "brand": brand, "riskType": risk_type, "barcode": barcode, "since": since, "until": until, "withMatches": with_matches, "limit": limit, "cursor": cursor}) def search_recall_matches(self, marketplace=None, brand=None, category=None, alert_id=None, min_confidence=None, since=None, limit=None, cursor=None): """Recall Radar: marketplace listings that match a recall alert (GET /v1/recalls/matches)""" return self._request("GET", "/v1/recalls/matches", {"marketplace": marketplace, "brand": brand, "category": category, "alertId": alert_id, "minConfidence": min_confidence, "since": since, "limit": limit, "cursor": cursor}) def get_job(self, id): """One job with its description (GET /v1/jobs/{id})""" return self._request("GET", "/v1/jobs/{id}".format(id=quote(str(id), safe="")), None) def resolve_company(self, domain=None, name=None, country=None, id=None, type=None): """EU Company ID: domain <-> registry id / VAT / LEI, legal name, country and stored register status (GET /v1/companies/resolve)""" return self._request("GET", "/v1/companies/resolve", {"domain": domain, "name": name, "country": country, "id": id, "type": type}) def verify_company(self, id=None, country=None, vat=None): """EU Company ID: live status check in the national register and EU VAT check in VIES (GET /v1/companies/verify)""" return self._request("GET", "/v1/companies/verify", {"id": id, "country": country, "vat": vat}) def get_company_profile(self, domain): """Company profile: job boards, open jobs, hiring signals, tech, events, Shopify store (GET /v1/companies/{domain})""" return self._request("GET", "/v1/companies/{domain}".format(domain=quote(str(domain), safe="")), None) def get_company_hiring_history(self, domain, from_=None, to=None): """Point-in-time hiring history 2023 onwards (open jobs per crawl, from the public web archive) (GET /v1/companies/{domain}/history)""" return self._request("GET", "/v1/companies/{domain}/history".format(domain=quote(str(domain), safe="")), {"from": from_, "to": to}) def get_ats_market(self, ats=None, country=None, size_band=None, series=None, from_=None, to=None): """ATS market share 2023 onwards: monthly share, wins and losses per vendor, plus today’s live share (GET /v1/ats/market)""" return self._request("GET", "/v1/ats/market", {"ats": ats, "country": country, "sizeBand": size_band, "series": series, "from": from_, "to": to}) def list_ats_switches(self, from_ats=None, to_ats=None, since=None, country=None, include_lever_gap=None, domain=None, limit=None, cursor=None): """Companies that switched ATS, dated (e.g. Lever -> Ashby), keyed by company domain (GET /v1/ats/switches)""" return self._request("GET", "/v1/ats/switches", {"fromAts": from_ats, "toAts": to_ats, "since": since, "country": country, "includeLeverGap": include_lever_gap, "domain": domain, "limit": limit, "cursor": cursor}) def get_company_pulse(self, domain, since=None, types=None): """Company Pulse: dated change events of one company from every dataset (last 90 days) (GET /v1/companies/{domain}/pulse)""" return self._request("GET", "/v1/companies/{domain}/pulse".format(domain=quote(str(domain), safe="")), {"since": since, "types": types}) def list_pulse_events(self, since=None, until=None, country=None, types=None, limit=None, cursor=None): """Company Pulse feed: events of every company, filtered by day, country and type (GET /v1/pulse/events)""" return self._request("GET", "/v1/pulse/events", {"since": since, "until": until, "country": country, "types": types, "limit": limit, "cursor": cursor}) def get_pay_bands(self, country=None, isco=None, role=None, week=None, limit=None): """Posted-pay bands (p25 / median / p75, EUR) per country and ISCO-08 group (GET /v1/pay/bands)""" return self._request("GET", "/v1/pay/bands", {"country": country, "isco": isco, "role": role, "week": week, "limit": limit}) def list_device_manufacturers(self, domain=None, srn=None, q=None, country=None, risk_class=None, us_cleared=None, no_eu_ar=None, cert_expiring=None, no_devices=None, limit=None, cursor=None): """Medical-device manufacturers in EUDAMED, with certificates and FDA cross-reference (GET /v1/devices/manufacturers)""" return self._request("GET", "/v1/devices/manufacturers", {"domain": domain, "srn": srn, "q": q, "country": country, "risk_class": risk_class, "us_cleared": us_cleared, "no_eu_ar": no_eu_ar, "cert_expiring": cert_expiring, "no_devices": no_devices, "limit": limit, "cursor": cursor}) def list_devices(self, manufacturer_domain=None, manufacturer_srn=None, risk_class=None, status=None, limit=None, cursor=None): """Registered medical devices (UDI-DI) of one manufacturer, with its certificates and FDA matches (GET /v1/devices)""" return self._request("GET", "/v1/devices", {"manufacturer_domain": manufacturer_domain, "manufacturer_srn": manufacturer_srn, "risk_class": risk_class, "status": status, "limit": limit, "cursor": cursor}) def list_device_events(self, srn=None, kind=None, since=None, limit=None): """Change events: new registrations, certificates issued / expiring / withdrawn (needs an API key) (GET /v1/devices/events)""" return self._request("GET", "/v1/devices/events", {"srn": srn, "kind": kind, "since": since, "limit": limit}) def list_pharma_sites(self, country=None, domain=None, registry_id=None, status=None, since=None, q=None, limit=None, cursor=None): """Pharma manufacturing and distribution sites: EU GMP/GDP status, licences, non-compliance and FDA record (GET /v1/pharma/sites)""" return self._request("GET", "/v1/pharma/sites", {"country": country, "domain": domain, "registry_id": registry_id, "status": status, "since": since, "q": q, "limit": limit, "cursor": cursor}) def get_pharma_site(self, id): """One pharma site: its EudraGMDP documents, non-compliance statements, FDA inspections and warning letters (GET /v1/pharma/sites/{id})""" return self._request("GET", "/v1/pharma/sites/{id}".format(id=quote(str(id), safe="")), None) def get_pay_transparency(self, country=None, size_band=None, since=None, until=None, limit=None): """Pay-transparency adoption: share of job ads stating pay, per country and day (GET /v1/pay/transparency)""" return self._request("GET", "/v1/pay/transparency", {"country": country, "sizeBand": size_band, "since": since, "until": until, "limit": limit}) def search_energy_projects(self, country=None, technology=None, min_mw=None, status=None, changed_since=None, region=None, operator_domain=None, limit=None, cursor=None): """Grid-connection and renewable project pipeline (DE + GB, 1 MW and up) (GET /v1/energy/projects)""" return self._request("GET", "/v1/energy/projects", {"country": country, "technology": technology, "minMw": min_mw, "status": status, "changedSince": changed_since, "region": region, "operatorDomain": operator_domain, "limit": limit, "cursor": cursor}) def get_energy_project(self, id): """One energy project with its change history (GET /v1/energy/projects/{id})""" return self._request("GET", "/v1/energy/projects/{id}".format(id=quote(str(id), safe="")), None) def search_emissions_companies(self, domain=None, lei=None, country=None, sector=None, limit=None, cursor=None): """Company-level emissions: EU ETS operators with LEI, website and parent (CBAM, financed emissions) (GET /v1/emissions/companies)""" return self._request("GET", "/v1/emissions/companies", {"domain": domain, "lei": lei, "country": country, "sector": sector, "limit": limit, "cursor": cursor}) def get_emissions_installation(self, id): """One EU ETS installation: yearly emissions, allocation, compliance, change history (GET /v1/emissions/installations/{id})""" return self._request("GET", "/v1/emissions/installations/{id}".format(id=quote(str(id), safe="")), None) def list_rate_curve_cities(self): """Cities with a forward hotel-rate curve (GET /v1/hotels/rate-curve)""" return self._request("GET", "/v1/hotels/rate-curve", None) def get_rate_curve(self, city, snapshot=None, from_=None, to=None): """Forward hotel-rate curve and demand index of a city (next 90 nights) (GET /v1/hotels/rate-curve/{city})""" return self._request("GET", "/v1/hotels/rate-curve/{city}".format(city=quote(str(city), safe="")), {"snapshot": snapshot, "from": from_, "to": to}) def get_shop(self, domain): """Shopify store profile on a domain (GET /v1/shops/{domain})""" return self._request("GET", "/v1/shops/{domain}".format(domain=quote(str(domain), safe="")), None) def get_shop_apps(self, domain): """Apps detected on a Shopify store, with installs and uninstalls (GET /v1/shops/{domain}/apps)""" return self._request("GET", "/v1/shops/{domain}/apps".format(domain=quote(str(domain), safe="")), None) def search_tenders(self, q=None, country=None, cpv=None, kind=None, source=None, published_since=None, published_until=None, open_on=None, min_value_eur=None, winner_domain=None, sector=None, limit=None, cursor=None): """Search public tenders and contract awards (EU TED, UK Find a Tender, FR/DE/PL/BG national boards) (GET /v1/tenders)""" return self._request("GET", "/v1/tenders", {"q": q, "country": country, "cpv": cpv, "kind": kind, "source": source, "publishedSince": published_since, "publishedUntil": published_until, "openOn": open_on, "minValueEur": min_value_eur, "winnerDomain": winner_domain, "sector": sector, "limit": limit, "cursor": cursor}) def list_expiring_contracts(self, country=None, cpv=None, from_=None, to=None, min_value_eur=None, sector=None, frameworks_only=None, kind=None, limit=None, cursor=None): """Re-tender Radar: public contracts that end in a window, with the current holder (GET /v1/tenders/expiring)""" return self._request("GET", "/v1/tenders/expiring", {"country": country, "cpv": cpv, "from": from_, "to": to, "minValueEur": min_value_eur, "sector": sector, "frameworksOnly": frameworks_only, "kind": kind, "limit": limit, "cursor": cursor}) def get_tender_supplier(self, id): """Contracts Intelligence: supplier profile (wins, buyers, CPVs, countries, timeline, concentration) (GET /v1/tenders/suppliers/{id})""" return self._request("GET", "/v1/tenders/suppliers/{id}".format(id=quote(str(id), safe="")), None) def get_tender_market(self, country=None, cpv=None, since=None, until=None): """Contracts Intelligence: one market (country x CPV): top buyers and winners, HHI, award vs estimate (GET /v1/tenders/markets)""" return self._request("GET", "/v1/tenders/markets", {"country": country, "cpv": cpv, "since": since, "until": until}) def get_tender_leaders(self, q=None, country=None, cpv=None, months=None, sector=None, limit=None): """Who wins public contracts: top winners and buyers of award notices for a CPV, keyword and country (GET /v1/tenders/leaders)""" return self._request("GET", "/v1/tenders/leaders", {"q": q, "country": country, "cpv": cpv, "months": months, "sector": sector, "limit": limit}) def list_trials(self, sponsor=None, sponsor_class=None, phase=None, status=None, condition=None, area=None, country=None, changed_since=None, limit=None, cursor=None): """Clinical trials by sponsor, phase, status, condition and country, merged across ClinicalTrials.gov and EU CTIS (GET /v1/trials)""" return self._request("GET", "/v1/trials", {"sponsor": sponsor, "sponsorClass": sponsor_class, "phase": phase, "status": status, "condition": condition, "area": area, "country": country, "changedSince": changed_since, "limit": limit, "cursor": cursor}) def get_trial_sponsor(self, domain): """Clinical-trial pipeline of one sponsor company: phases, statuses, areas, countries, recent changes, hiring and tenders (GET /v1/trials/sponsors/{domain})""" return self._request("GET", "/v1/trials/sponsors/{domain}".format(domain=quote(str(domain), safe="")), None) def list_merger_filings(self, authority=None, country=None, sector=None, nace=None, status=None, since=None, party_domain=None, party_registry_id=None, party=None, limit=None, cursor=None): """Merger-control filings: notified concentrations at the EU Commission, Bundeskartellamt and UK CMA (GET /v1/mergers)""" return self._request("GET", "/v1/mergers", {"authority": authority, "country": country, "sector": sector, "nace": nace, "status": status, "since": since, "party_domain": party_domain, "party_registry_id": party_registry_id, "party": party, "limit": limit, "cursor": cursor}) def get_merger_filing(self, id): """One merger case with its parties and dated events (notified, phase 2, decisions) (GET /v1/mergers/{id})""" return self._request("GET", "/v1/mergers/{id}".format(id=quote(str(id), safe="")), None) def list_planning_applications(self, country=None, authority=None, type=None, kind=None, status=None, min_units=None, min_floor_area=None, changed_since=None, with_applicant=None, limit=None, cursor=None): """UK and Ireland construction leads: commercial and large residential planning applications (GET /v1/planning/applications)""" return self._request("GET", "/v1/planning/applications", {"country": country, "authority": authority, "type": type, "kind": kind, "status": status, "min_units": min_units, "min_floor_area": min_floor_area, "changed_since": changed_since, "with_applicant": with_applicant, "limit": limit, "cursor": cursor}) def get_planning_company(self, domain): """Planning activity of one developer or builder, with its registry id and construction tender wins (GET /v1/planning/companies/{domain})""" return self._request("GET", "/v1/planning/companies/{domain}".format(domain=quote(str(domain), safe="")), None) def get_consent_market(self, country=None, segment=None, month=None): """Consent-management platform (CMP) market share and tracker prevalence, by country or segment (GET /v1/consent/market)""" return self._request("GET", "/v1/consent/market", {"country": country, "segment": segment, "month": month}) def get_consent_company(self, domain): """Which CMP and which trackers one company website loads, with its history (GET /v1/consent/companies/{domain})""" return self._request("GET", "/v1/consent/companies/{domain}".format(domain=quote(str(domain), safe="")), None) def list_consent_switches(self, domain=None, from_cmp=None, to_cmp=None, country=None, kind=None, since=None, limit=None, cursor=None): """Companies that switched, added or dropped a consent-management platform (GET /v1/consent/switches)""" return self._request("GET", "/v1/consent/switches", {"domain": domain, "from_cmp": from_cmp, "to_cmp": to_cmp, "country": country, "kind": kind, "since": since, "limit": limit, "cursor": cursor}) def check_gambling_domain(self, domain): """Is this website licensed for gambling, where, by whom, and with which licence status (GET /v1/gambling/domains/{domain})""" return self._request("GET", "/v1/gambling/domains/{domain}".format(domain=quote(str(domain), safe="")), None) def bulk_check_gambling_domains(self, body): """Bulk licence check: up to 100 domains in one call (POST /v1/gambling/domains/check)""" return self._request("POST", "/v1/gambling/domains/check", None, body) def list_gambling_operators(self, q=None, regulator=None, country=None, include_removed=None, limit=None, cursor=None): """Licensed gambling operators per market, with licences, domains and registry ids (GET /v1/gambling/operators)""" return self._request("GET", "/v1/gambling/operators", {"q": q, "regulator": regulator, "country": country, "includeRemoved": include_removed, "limit": limit, "cursor": cursor}) def list_gambling_events(self, regulator=None, country=None, type=None, since=None, limit=None, cursor=None): """Change feed: licences granted, suspended, revoked or ended, domains added or removed (GET /v1/gambling/events)""" return self._request("GET", "/v1/gambling/events", {"regulator": regulator, "country": country, "type": type, "since": since, "limit": limit, "cursor": cursor}) def create_watchlist(self, body): """Track a list of companies (needs an API key) (POST /v1/watchlists)""" return self._request("POST", "/v1/watchlists", None, body) def list_watchlists(self): """Your watchlists (GET /v1/watchlists)""" return self._request("GET", "/v1/watchlists", None) def get_watchlist(self, id): """One watchlist with its domains (GET /v1/watchlists/{id})""" return self._request("GET", "/v1/watchlists/{id}".format(id=quote(str(id), safe="")), None) def delete_watchlist(self, id): """Stop tracking a watchlist (DELETE /v1/watchlists/{id})""" return self._request("DELETE", "/v1/watchlists/{id}".format(id=quote(str(id), safe="")), None) def get_watchlist_changes(self, id, since=None, types=None, limit=None): """Company Pulse events and the latest hiring signals of the tracked companies (GET /v1/watchlists/{id}/changes)""" return self._request("GET", "/v1/watchlists/{id}/changes".format(id=quote(str(id), safe="")), {"since": since, "types": types, "limit": limit}) def list_plans(self): """Plans, limits and prices (GET /v1/plans)""" return self._request("GET", "/v1/plans", None) def get_me(self): """Your tier, plan, quota and usage (GET /v1/me)""" return self._request("GET", "/v1/me", None) def signup(self, body): """Sign up: email a one-time link that shows a free API key (POST /v1/account/signup)""" return self._request("POST", "/v1/account/signup", None, body) def verify_signup(self, body): """Redeem the emailed link: returns the API key (once) (POST /v1/account/verify)""" return self._request("POST", "/v1/account/verify", None, body) def create_checkout(self, body): """Upgrade: a Stripe Checkout link for Starter, Pro or Business (POST /v1/account/checkout)""" return self._request("POST", "/v1/account/checkout", None, body) def export_account_data(self): """Download everything we hold about your account (GDPR access and portability) (GET /v1/account/export)""" return self._request("GET", "/v1/account/export", None) def delete_account(self, body): """Delete the account and erase the email address (GDPR erasure) (DELETE /v1/account)""" return self._request("DELETE", "/v1/account", None, body) def get_billing_portal(self): """Billing portal link: invoices, card, plan change, cancel (GET /v1/account/portal)""" return self._request("GET", "/v1/account/portal", None) def watch_competitors(self, body): """Watch competitors: a weekly email digest per competitor domain (POST /v1/watch/competitors)""" return self._request("POST", "/v1/watch/competitors", None, body) def list_competitor_watches(self): """Your watched competitors, limit and free month (GET /v1/watch/competitors)""" return self._request("GET", "/v1/watch/competitors", None) def unwatch_competitor(self, domain): """Stop watching one competitor (DELETE /v1/watch/competitors/{domain})""" return self._request("DELETE", "/v1/watch/competitors/{domain}".format(domain=quote(str(domain), safe="")), None) def create_addon_checkout(self, body): """Buy Competitor Watch: a Stripe Checkout link for the add-on (POST /v1/watch/checkout)""" return self._request("POST", "/v1/watch/checkout", None, body) def unsubscribe_digests(self, token): """One-click unsubscribe from the Competitor Watch digests (link in every digest) (POST /v1/watch/unsubscribe)""" return self._request("POST", "/v1/watch/unsubscribe", {"token": token}) def watch_tenders(self, body): """Save a tender search: a weekly email with its new awards and open calls (POST /v1/watch/tenders)""" return self._request("POST", "/v1/watch/tenders", None, body) def list_tender_watches(self): """Your saved tender searches (GET /v1/watch/tenders)""" return self._request("GET", "/v1/watch/tenders", None) def unwatch_tenders(self, id): """Delete one saved tender search (DELETE /v1/watch/tenders/{id})""" return self._request("DELETE", "/v1/watch/tenders/{id}".format(id=quote(str(id), safe="")), None) def watch_jobs(self, body): """Save a remote-job search: an email with its new jobs, daily or weekly (POST /v1/watch/jobs)""" return self._request("POST", "/v1/watch/jobs", None, body) def list_job_watches(self): """Your saved remote-job searches (GET /v1/watch/jobs)""" return self._request("GET", "/v1/watch/jobs", None) def unwatch_jobs(self, id): """Delete one saved remote-job search (DELETE /v1/watch/jobs/{id})""" return self._request("DELETE", "/v1/watch/jobs/{id}".format(id=quote(str(id), safe="")), None) def watch_recalls(self, body): """Watch a brand: a weekly email with its new product safety alerts and marketplace matches (POST /v1/watch/recalls)""" return self._request("POST", "/v1/watch/recalls", None, body) def list_recall_watches(self): """Your watched brands (GET /v1/watch/recalls)""" return self._request("GET", "/v1/watch/recalls", None) def unwatch_recalls(self, id): """Stop watching one brand (DELETE /v1/watch/recalls/{id})""" return self._request("DELETE", "/v1/watch/recalls/{id}".format(id=quote(str(id), safe="")), None) def watch_devices(self, body): """Follow an EUDAMED manufacturer: an email when it registers devices or a certificate changes (POST /v1/watch/devices)""" return self._request("POST", "/v1/watch/devices", None, body) def list_device_watches(self): """Your followed EUDAMED manufacturers (GET /v1/watch/devices)""" return self._request("GET", "/v1/watch/devices", None) def watch_devices_bulk(self, body): """Follow many EUDAMED manufacturers at once from a CSV of SRNs (client portfolio) (POST /v1/watch/devices/bulk)""" return self._request("POST", "/v1/watch/devices/bulk", None, body) def unwatch_devices(self, id): """Stop following one manufacturer (DELETE /v1/watch/devices/{id})""" return self._request("DELETE", "/v1/watch/devices/{id}".format(id=quote(str(id), safe="")), None) def watch_mergers(self, body): """Follow merger filings: a daily email when a new filing or a status change matches (POST /v1/watch/mergers)""" return self._request("POST", "/v1/watch/mergers", None, body) def list_merger_watches(self): """Your Merger Watch rules and allowance (GET /v1/watch/mergers)""" return self._request("GET", "/v1/watch/mergers", None) def unwatch_mergers(self, id): """Remove one Merger Watch rule (DELETE /v1/watch/mergers/{id})""" return self._request("DELETE", "/v1/watch/mergers/{id}".format(id=quote(str(id), safe="")), None) def watch_suppliers(self, body): """Upload a supplier list: a risk digest of status, VAT and contract changes by email (POST /v1/watch/suppliers)""" return self._request("POST", "/v1/watch/suppliers", None, body) def list_supplier_watches(self, limit=None, offset=None): """The suppliers you follow, plan limit and digest schedule (GET /v1/watch/suppliers)""" return self._request("GET", "/v1/watch/suppliers", {"limit": limit, "offset": offset}) def unwatch_supplier(self, id): """Stop following one supplier (DELETE /v1/watch/suppliers/{id})""" return self._request("DELETE", "/v1/watch/suppliers/{id}".format(id=quote(str(id), safe="")), None) def supplier_risk(self, domains, days=None): """Supplier risk: register, VAT, accounts, contract and hiring changes at a supplier domain, ranked (GET /v1/suppliers/risk)""" return self._request("GET", "/v1/suppliers/risk", {"domains": domains, "days": days}) def check_coverage(self, body): """Which of up to 50 company domains have a known public job board (POST /v1/coverage)""" return self._request("POST", "/v1/coverage", None, body) def get_company_board(self, domain): """Primary job board and open-jobs snapshot of a company (prefer /v1/companies/{domain}) (GET /v1/company/{domain})""" return self._request("GET", "/v1/company/{domain}".format(domain=quote(str(domain), safe="")), None) def get_company_signals(self, domain): """Latest daily hiring signals of a company (GET /v1/company/{domain}/signals)""" return self._request("GET", "/v1/company/{domain}/signals".format(domain=quote(str(domain), safe="")), None) def list_hiring_events(self, type=None, since=None, country=None, limit=None, provisional=None): """Hiring events feed (market entry, ATS switch, surge, freeze, new department) (GET /v1/events)""" return self._request("GET", "/v1/events", {"type": type, "since": since, "country": country, "limit": limit, "provisional": provisional}) def get_review_snapshot(self, q, platform=None, days=None, limit=None): """Rating trajectory of tracked review listings (GET /v1/reviews/snapshot)""" return self._request("GET", "/v1/reviews/snapshot", {"q": q, "platform": platform, "days": days, "limit": limit}) def get_review_panel(self, ticker=None, q=None, platform=None, since=None, until=None, rollup=None, limit=None): """Daily review velocity panel by ticker or name (derived metrics only) (GET /v1/reviews/panel)""" return self._request("GET", "/v1/reviews/panel", {"ticker": ticker, "q": q, "platform": platform, "since": since, "until": until, "rollup": rollup, "limit": limit})