twitterapi.compliance.compliance module

compliance.py: Implementation of Twitter batch compliance.

“These batch compliance endpoints allow you to upload large datasets of Tweet or user IDs to retrieve their compliance status in order to determine what data requires action in order to bring your datasets into compliance. Please note, use of the batch compliance endpoints is restricted to aforementioned use cases, and any other purpose is prohibited and may result in enforcement action.” (Source: https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/introduction´)

Working with the batch compliance endpoints generally involves 5 steps:

  1. Preparing the list of Tweet IDs or user IDs

  2. Creating a compliance job

  3. Uploading the list of Tweet IDs or user IDs

  4. Checking the status of the compliance job

  5. Downloading the results

More detailed information can be found at: https://developer.twitter.com/en/docs/twitter-api/compliance/batch-compliance/introduction

Examples

Creating a compliance job:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance import create_compliance_job
from sparta.twitterapi.models.twitter_v2_spec import ComplianceJobType

compliancejobresponse = await create_compliance_job(ComplianceJobType.tweets, 'testjob')
if not compliancejobresponse.data:
    raise Exception()
else:
    compliancejob = compliancejobresponse.data
jobid = compliancejob.id
uploadurl = compliancejob.upload_url

Uploading the list of Tweet IDs or user IDs:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance import upload_ids

await upload_ids(uploadurl, "filename.txt")

Checking the status of the compliance job:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance import list_job
from sparta.twitterapi.models.twitter_v2_spec import ComplianceJobStatus

async def check_job(jobid: str) -> bool:
    compliancejobresponse = await list_job(jobid)
    if not compliancejobresponse.data:
        raise Exception()
    else:
        compliancejob = compliancejobresponse.data

    if compliancejob.status == ComplianceJobStatus.complete:
        return True
    else:
        return False

await check_job(jobid)

Downloading the results:

import os
os.environ["BEARER_TOKEN"] = "xxxxxxxxxxx"
from sparta.twitterapi.compliance.compliance import download_results, list_job

compliancejobresponse = await list_job(jobid)
if not compliancejobresponse.data:
    raise Exception()
else:
    compliancejob = compliancejobresponse.data
downloadurl = compliancejob.download_url

await download_results(downloadurl, 'results.txt')
async sparta.twitterapi.compliance.compliance.create_compliance_job(type: ComplianceJobType, name: str, resumable: bool = False) → CreateComplianceJobResponse

Creates a compliance job.

Parameters:

  • type (str) – Type of Compliance Job to list. Possible values = tweets, users

  • name (str) – User-provided name for a compliance job.

  • resumable (bool) – If true, this endpoint will return a pre-signed URL with resumable uploads enabled.

Raises:

Exception – Compliance Job can not be created due to an http error.

Returns:

Returns an Twitter CreateComplianceJobResponse object.

Return type:

CreateComplianceJobResponse

async sparta.twitterapi.compliance.compliance.download_results(download_url: Url, results_file_path: str) → int

Downloads the results of a compliance job and saves them to a file.

Parameters:

  • download_url (AnyUrl) – Download URL for a Jompliance Job

  • results_file_path (str) – File path where the results will be saved.

Raises:

  • Exception – Results can not be fetched due to an http error.

  • Exception – Results can not be writen into the file.

Returns:

Respone status.

Return type:

int

async sparta.twitterapi.compliance.compliance.list_job(id: str) → Get2ComplianceJobsIdResponse

Returns a single Compliance Job by ID.

Parameters:

id (str) – The ID of the Compliance Job to retrieve.

Raises:

Exception – Compliance Job can not be fetched due to an http error.

Returns:

Returns an Twitter Get2ComplianceJobsIdResponse object.

Return type:

Get2ComplianceJobsIdResponse

async sparta.twitterapi.compliance.compliance.list_jobs(type: ComplianceJobType, status: ComplianceJobStatus | None = None) → Get2ComplianceJobsResponse

Returns recent Compliance Jobs for a given job type and optional job status.

Parameters:

  • type (str) – Type of Compliance Job to list. Possible values = tweets, users

  • status (str, optional) – Status of Compliance Job to list. Possible values = created, in_progress, failed, complete. Defaults to None.

Raises:

Exception – Compliance Jobs can not be fetched due to an http error.

Returns:

Returns an Twitter Get2ComplianceJobsResponse object.

Return type:

Get2ComplianceJobsResponse

async sparta.twitterapi.compliance.compliance.upload_ids(upload_url: Url, ids_file_path: str) → str

Uploads IDs for a compliance job.

Parameters:

  • upload_url (AnyUrl) – Upload URL for a Compliance Job

  • ids_file_path (str) – File path where the results IDs are in. Plain text file newline separated.

Raises:

Exception – IDs cannot uploaded due to an http error.

Returns:

Response text

Return type:

str