MarqetiveLib
The async Python client library behind Postey - the social media publishing platform. Simple, type-safe, and async-ready.
Supported Platforms
All nine platforms Postey publishes to:
- X (Twitter) - Post tweets, upload media, manage threads
- LinkedIn - Share updates, upload images and videos
- Instagram - Create posts with media via Graph API
- TikTok - Upload and publish videos
- YouTube - Upload Shorts and videos
- Threads - Publish text and media posts via Threads Graph API
- Bluesky - Post via the AT Protocol
- Facebook - Publish to Pages via Graph API
- Pinterest - Create pins with media
Features
- Unified API: Single interface for all social media platforms
- Async-First: Built for modern async Python applications
- Type-Safe: Full type hints and Pyright compliance
- Auto Token Refresh: Factory handles OAuth token lifecycle automatically
- Media Upload: Progress tracking for large file uploads
- Retry Logic: Exponential backoff with jitter for transient failures
- Well Tested: Comprehensive test coverage
Installation
pip install marqetive-lib
Or with Poetry:
poetry add marqetive-lib
Quick Start
import asyncio
from marqetive import get_client, AuthCredentials, PostCreateRequest
async def main():
# Create credentials for your platform
credentials = AuthCredentials(
platform="twitter",
access_token="your_access_token",
refresh_token="your_refresh_token"
)
# Get authenticated client (auto-refreshes token if expired)
client = await get_client(credentials)
# Use client as async context manager
async with client:
request = PostCreateRequest(content="Hello from MarqetiveLib!")
post = await client.create_post(request)
print(f"Posted! ID: {post.platform_id}")
asyncio.run(main())
Platform Examples
from marqetive import get_client, AuthCredentials, PostCreateRequest
credentials = AuthCredentials(
platform="twitter",
access_token="your_access_token",
refresh_token="your_refresh_token" # Optional, for token refresh
)
client = await get_client(credentials)
async with client:
# Text post
post = await client.create_post(PostCreateRequest(content="Hello Twitter!"))
# Post with media
post = await client.create_post(PostCreateRequest(
content="Check out this image!",
media_paths=["/path/to/image.jpg"]
))
from marqetive import get_client, AuthCredentials, PostCreateRequest
credentials = AuthCredentials(
platform="linkedin",
access_token="your_access_token",
user_id="urn:li:person:your_person_id" # Required for LinkedIn
)
client = await get_client(credentials)
async with client:
post = await client.create_post(PostCreateRequest(
content="Excited to share this update!"
))
from marqetive import get_client, AuthCredentials, PostCreateRequest
credentials = AuthCredentials(
platform="instagram",
access_token="your_access_token",
user_id="your_instagram_business_account_id" # Required
)
client = await get_client(credentials)
async with client:
# Instagram requires media for posts
post = await client.create_post(PostCreateRequest(
content="Beautiful day! #photography",
media_paths=["/path/to/photo.jpg"]
))
TikTok
from marqetive import get_client, AuthCredentials, PostCreateRequest
credentials = AuthCredentials(
platform="tiktok",
access_token="your_access_token",
additional_data={"open_id": "your_open_id"} # Required for TikTok
)
client = await get_client(credentials)
async with client:
post = await client.create_post(PostCreateRequest(
content="Check out this video!",
media_paths=["/path/to/video.mp4"]
))
Using Custom OAuth Credentials
from marqetive import PlatformFactory, AuthCredentials, PostCreateRequest
# Create factory with your OAuth app credentials
factory = PlatformFactory(
twitter_client_id="your_client_id",
twitter_client_secret="your_client_secret",
linkedin_client_id="your_linkedin_client_id",
linkedin_client_secret="your_linkedin_client_secret"
)
credentials = AuthCredentials(
platform="twitter",
access_token="user_access_token",
refresh_token="user_refresh_token"
)
# Get client with automatic token refresh
client = await factory.get_client(credentials)
async with client:
post = await client.create_post(PostCreateRequest(content="Hello!"))
Progress Tracking for Media Uploads
def progress_callback(operation: str, progress: int, total: int, message: str | None):
percent = (progress / total) * 100 if total > 0 else 0
print(f"{operation}: {percent:.1f}% - {message or ''}")
client = await get_client(credentials, progress_callback=progress_callback)
async with client:
post = await client.create_post(PostCreateRequest(
content="Uploading video...",
media_paths=["/path/to/large_video.mp4"]
))
Error Handling
from marqetive import get_client, AuthCredentials, PostCreateRequest
from marqetive.core.exceptions import (
PlatformError,
PlatformAuthError,
RateLimitError,
MediaUploadError
)
try:
client = await get_client(credentials)
async with client:
post = await client.create_post(request)
except PlatformAuthError as e:
print(f"Authentication failed: {e}")
# Token may need refresh or reconnection
except RateLimitError as e:
print(f"Rate limited. Retry after: {e.retry_after} seconds")
except MediaUploadError as e:
print(f"Media upload failed: {e}")
except PlatformError as e:
print(f"Platform error: {e}")
Development
Setup Development Environment
# Clone the repository
git clone git@github.com:quantplay-admin/MarqetiveLib.git
cd MarqetiveLib
# Install Poetry if you haven't already
curl -sSL https://install.python-poetry.org | python3 -
# Install dependencies
poetry install --with dev,docs
# Activate virtual environment
poetry shell
Running Tests
The tests/usage money-path suites need a live Redis and fail without one — they are
the only coverage of the debit Lua and the fair-use ceiling, so they must never skip:
docker run --rm -p 6379:6379 redis:7-alpine
Set REDIS_URL to point elsewhere (default redis://localhost:6379). The suite flushes
DB index REDIS_TEST_DB (default 15) around every test.
# Run tests
poetry run pytest
# Run tests with coverage
poetry run pytest --cov=src/marqetive --cov-report=term-missing
# Run platform-specific tests
poetry run pytest tests/platforms/test_twitter.py
Code Quality
# Lint code with Ruff
poetry run ruff check .
# Format code with Ruff
poetry run ruff format .
# Type check with Pyright
poetry run pyright src/
License
This project is licensed under the MIT License - see the LICENSE file for details.
Metadata
Release files for marqetive-lib 0.33.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| marqetive_lib-0.33.0.tar.gz | 351.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| marqetive_lib-0.33.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 769.7 kB
Release files / marqetive_lib-0.33.0.tar.gz
| Download URL | marqetive_lib-0.33.0.tar.gz |
|---|---|
| Size | 351.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
e87ef1698acf87b7aa880cc04bc0e0d361c7dea1b5c53e5efcdceb321492544b
|
|
BLAKE2b-256 checksum How to use checksums |
f7532363b3f4e8b693011f68e42429e2239a3f861ed975c6e7b00d139a4be405
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.4.1 CPython/3.14.7 Darwin/24.6.0
|
Release files / marqetive_lib-0.33.0-py3-none-any.whl
| Download URL | marqetive_lib-0.33.0-py3-none-any.whl |
|---|---|
| Size | 418.7 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
779423c8b913fa56825fb32f982dc20f6b14cd6341a621e1657c79b88df96d1f
|
|
BLAKE2b-256 checksum How to use checksums |
00fc4177b8c5d084b1d09ce974ed1301493d0ea4534669f380ab91bddeed21fe
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
poetry/2.4.1 CPython/3.14.7 Darwin/24.6.0
|