# LastCron
[](https://badge.fury.io/py/lastcron)
[](https://pypi.org/project/lastcron/)
[](LICENSE)
[](https://github.com/allanbru/lastcron-sdk/actions/workflows/ci.yml)
**LastCron** is a Python SDK for building and orchestrating workflows with the LastCron platform. It provides a simple, decorator-based API for creating flows that can be scheduled, monitored, and managed through a centralized orchestration system.
## ✨ Features
- **🎯 Simple Decorator API**: Mark functions as flows with a single `@flow` decorator
- **🔄 Flow Orchestration**: Trigger and chain flows programmatically
- **📊 Automatic Logging**: Built-in logger that sends logs to the orchestrator
- **🔐 Secret Management**: Secure handling of secrets with automatic redaction
- **⏰ Scheduling**: Schedule flows for immediate or future execution
- **🔍 Type Safety**: Strongly-typed dataclasses for all API responses
- **⚡ Async Support**: Both synchronous and asynchronous API clients
- **📦 On-Demand Configuration**: Fetch configuration blocks only when needed
## 📦 Installation
Install LastCron using pip:
```bash
pip install lastcron
```
Or with optional development dependencies:
```bash
pip install lastcron[dev]
```
## 🚀 Quick Start
### Basic Flow
Create a simple flow that logs messages:
```python
from lastcron import flow, get_run_logger, get_workspace_id
@flow
def hello_world(**params):
logger = get_run_logger()
workspace_id = get_workspace_id()
logger.info(f"Hello from workspace {workspace_id}!")
logger.info(f"Parameters: {params}")
```
### Triggering Other Flows
Chain flows together by triggering them programmatically:
```python
from lastcron import flow, get_run_logger
@flow
def data_processing(**params):
logger = get_run_logger()
logger.info("Processing data...")
# Your data processing logic here
@flow
def orchestrator(**params):
logger = get_run_logger()
# Trigger another flow using .submit()
run = data_processing.submit(
parameters={'batch_size': 100}
)
if run:
logger.info(f"Triggered run ID: {run.id}")
```
### Using Configuration Blocks
Retrieve configuration and secrets on-demand:
```python
from lastcron import flow, get_block, get_run_logger
@flow
def api_integration(**params):
logger = get_run_logger()
# Get API credentials (automatically decrypted if secret)
api_key = get_block('api-key')
if api_key:
logger.info("Retrieved API credentials")
# The secret value is automatically redacted from logs
# Use api_key.value for the actual key
```
### Scheduling Flows
Schedule flows for future execution:
```python
from lastcron import flow, run_flow, get_run_logger
from datetime import datetime, timedelta
@flow
def scheduler(**params):
logger = get_run_logger()
# Schedule a flow to run in 1 hour
future_time = datetime.now() + timedelta(hours=1)
run = run_flow(
'cleanup_job',
parameters={'task': 'cleanup'},
scheduled_start=future_time
)
if run:
logger.info(f"Flow scheduled for {future_time}")
```
## 📚 Core Concepts
### The `@flow` Decorator
The `@flow` decorator is the main entry point for creating orchestrated workflows. It:
- Manages the flow lifecycle (status updates)
- Provides automatic logging
- Handles errors and reports them to the orchestrator
- Enables auto-execution when run directly
### Flow Context Functions
Access flow context using these helper functions:
- `get_run_logger()` - Get the logger instance
- `get_workspace_id()` - Get the current workspace ID
- `get_block(key_name)` - Retrieve a configuration block
- `run_flow(flow_name, ...)` - Trigger another flow
### Flow Triggering
Trigger flows in two ways:
1. **Using `.submit()` method** (recommended):
```python
run = my_flow.submit(parameters={'key': 'value'})
```
2. **Using `run_flow()` function**:
```python
run = run_flow('my_flow', parameters={'key': 'value'})
```
## 🔐 Secret Management
LastCron automatically redacts secret values from logs:
```python
from lastcron import flow, get_block, get_run_logger
@flow
def secure_flow(**params):
logger = get_run_logger()
# Get a secret block
db_password = get_block('database-password')
if db_password:
# This will be redacted in logs: "Password: ****"
logger.info(f"Password: {db_password.value}")
# Use the actual value in your code
connect_to_database(password=db_password.value)
```
## 🔄 Advanced Usage
### Asynchronous API Client
For high-performance applications, use the async client:
```python
from lastcron import AsyncAPIClient
import asyncio
async def main():
async with AsyncAPIClient(token="your_token", base_url="http://localhost/api") as client:
# Trigger multiple flows concurrently
tasks = [
client.trigger_flow_by_name(1, "flow_a"),
client.trigger_flow_by_name(1, "flow_b"),
client.trigger_flow_by_name(1, "flow_c")
]
results = await asyncio.gather(*tasks)
asyncio.run(main())
```
### Type-Safe Data Classes
Use strongly-typed dataclasses for better IDE support:
```python
from lastcron import flow, get_run_logger, FlowRun, Block, BlockType
@flow
def typed_flow(**params):
logger = get_run_logger()
# Trigger a flow and get a typed response
run: FlowRun = my_other_flow.submit(parameters={'key': 'value'})
if run:
logger.info(f"Run ID: {run.id}")
logger.info(f"State: {run.state.value}")
# Get a block with type information
config: Block = get_block('config')
if config and config.type == BlockType.JSON:
logger.info("Got JSON configuration")
```
## 🛠️ Development
### Setting Up Development Environment
```bash
# Clone the repository
git clone https://github.com/allanbru/lastcron-sdk.git
cd lastcron-sdk
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install in development mode
pip install -e ".[dev]"
```
### Running Tests
```bash
# Run all tests
pytest
# Run with coverage
pytest --cov=lastcron --cov-report=html
# Run specific test file
pytest tests/test_flow.py
```
### Code Quality
```bash
# Format code
black lastcron/
# Lint code
ruff check lastcron/
# Type check
mypy lastcron/
```
## 📖 Documentation
For detailed documentation, see:
- [SDK Documentation](lastcron/README.md) - Comprehensive SDK guide
- [Contributing Guide](CONTRIBUTING.md) - How to contribute
- [Examples](examples/) - Example flows and use cases
## 🤝 Contributing
We welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details on:
- Setting up your development environment
- Code style and standards
- Testing requirements
- Pull request process
## 📄 License
This project is licensed under the **Elastic License 2.0** - see the [LICENSE](LICENSE) file for details.
The Elastic License 2.0 is a source-available license that allows you to:
- Use the software freely
- Modify and distribute the software
- Use it in commercial applications
With the limitation that you cannot provide the software as a managed service.
## 🙏 Acknowledgments
Built with ❤️ by AllanBR Creations
## 📞 Support
- **Issues**: [GitHub Issues](https://github.com/allanbru/lastcron-sdk/issues)
- **Discussions**: [GitHub Discussions](https://github.com/allanbru/lastcron-sdk/discussions)
## 🗺️ Roadmap
- [ ] Enhanced error handling and retry mechanisms
- [ ] Flow dependencies and DAG support
- [ ] Real-time flow monitoring
- [ ] Flow templates and reusable components
- [ ] Integration with popular data tools
- [ ] Web UI for flow visualization
---
**Note**: This SDK is designed to work with the LastCron orchestration platform. You'll need a running LastCron instance to execute flows.
Raw data
{
"_id": null,
"home_page": null,
"name": "lastcron",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "AllanBR Creations <allan@allanbr.net>",
"keywords": "orchestration, workflow, cron, task-management, automation, scheduling, data-pipeline, etl",
"author": null,
"author_email": "AllanBR Creations <allan@allanbr.net>",
"download_url": "https://files.pythonhosted.org/packages/a3/93/1494b5fa76607ac43f58dd7d106a4fbc6064ec8f813882b9c52c3985f7f3/lastcron-0.1.0.tar.gz",
"platform": null,
"description": "# LastCron\n\n[](https://badge.fury.io/py/lastcron)\n[](https://pypi.org/project/lastcron/)\n[](LICENSE)\n[](https://github.com/allanbru/lastcron-sdk/actions/workflows/ci.yml)\n\n**LastCron** is a Python SDK for building and orchestrating workflows with the LastCron platform. It provides a simple, decorator-based API for creating flows that can be scheduled, monitored, and managed through a centralized orchestration system.\n\n## \u2728 Features\n\n- **\ud83c\udfaf Simple Decorator API**: Mark functions as flows with a single `@flow` decorator\n- **\ud83d\udd04 Flow Orchestration**: Trigger and chain flows programmatically\n- **\ud83d\udcca Automatic Logging**: Built-in logger that sends logs to the orchestrator\n- **\ud83d\udd10 Secret Management**: Secure handling of secrets with automatic redaction\n- **\u23f0 Scheduling**: Schedule flows for immediate or future execution\n- **\ud83d\udd0d Type Safety**: Strongly-typed dataclasses for all API responses\n- **\u26a1 Async Support**: Both synchronous and asynchronous API clients\n- **\ud83d\udce6 On-Demand Configuration**: Fetch configuration blocks only when needed\n\n## \ud83d\udce6 Installation\n\nInstall LastCron using pip:\n\n```bash\npip install lastcron\n```\n\nOr with optional development dependencies:\n\n```bash\npip install lastcron[dev]\n```\n\n## \ud83d\ude80 Quick Start\n\n### Basic Flow\n\nCreate a simple flow that logs messages:\n\n```python\nfrom lastcron import flow, get_run_logger, get_workspace_id\n\n@flow\ndef hello_world(**params):\n logger = get_run_logger()\n workspace_id = get_workspace_id()\n \n logger.info(f\"Hello from workspace {workspace_id}!\")\n logger.info(f\"Parameters: {params}\")\n```\n\n### Triggering Other Flows\n\nChain flows together by triggering them programmatically:\n\n```python\nfrom lastcron import flow, get_run_logger\n\n@flow\ndef data_processing(**params):\n logger = get_run_logger()\n logger.info(\"Processing data...\")\n # Your data processing logic here\n\n@flow\ndef orchestrator(**params):\n logger = get_run_logger()\n \n # Trigger another flow using .submit()\n run = data_processing.submit(\n parameters={'batch_size': 100}\n )\n \n if run:\n logger.info(f\"Triggered run ID: {run.id}\")\n```\n\n### Using Configuration Blocks\n\nRetrieve configuration and secrets on-demand:\n\n```python\nfrom lastcron import flow, get_block, get_run_logger\n\n@flow\ndef api_integration(**params):\n logger = get_run_logger()\n \n # Get API credentials (automatically decrypted if secret)\n api_key = get_block('api-key')\n \n if api_key:\n logger.info(\"Retrieved API credentials\")\n # The secret value is automatically redacted from logs\n # Use api_key.value for the actual key\n```\n\n### Scheduling Flows\n\nSchedule flows for future execution:\n\n```python\nfrom lastcron import flow, run_flow, get_run_logger\nfrom datetime import datetime, timedelta\n\n@flow\ndef scheduler(**params):\n logger = get_run_logger()\n \n # Schedule a flow to run in 1 hour\n future_time = datetime.now() + timedelta(hours=1)\n \n run = run_flow(\n 'cleanup_job',\n parameters={'task': 'cleanup'},\n scheduled_start=future_time\n )\n \n if run:\n logger.info(f\"Flow scheduled for {future_time}\")\n```\n\n## \ud83d\udcda Core Concepts\n\n### The `@flow` Decorator\n\nThe `@flow` decorator is the main entry point for creating orchestrated workflows. It:\n\n- Manages the flow lifecycle (status updates)\n- Provides automatic logging\n- Handles errors and reports them to the orchestrator\n- Enables auto-execution when run directly\n\n### Flow Context Functions\n\nAccess flow context using these helper functions:\n\n- `get_run_logger()` - Get the logger instance\n- `get_workspace_id()` - Get the current workspace ID\n- `get_block(key_name)` - Retrieve a configuration block\n- `run_flow(flow_name, ...)` - Trigger another flow\n\n### Flow Triggering\n\nTrigger flows in two ways:\n\n1. **Using `.submit()` method** (recommended):\n ```python\n run = my_flow.submit(parameters={'key': 'value'})\n ```\n\n2. **Using `run_flow()` function**:\n ```python\n run = run_flow('my_flow', parameters={'key': 'value'})\n ```\n\n## \ud83d\udd10 Secret Management\n\nLastCron automatically redacts secret values from logs:\n\n```python\nfrom lastcron import flow, get_block, get_run_logger\n\n@flow\ndef secure_flow(**params):\n logger = get_run_logger()\n \n # Get a secret block\n db_password = get_block('database-password')\n \n if db_password:\n # This will be redacted in logs: \"Password: ****\"\n logger.info(f\"Password: {db_password.value}\")\n \n # Use the actual value in your code\n connect_to_database(password=db_password.value)\n```\n\n## \ud83d\udd04 Advanced Usage\n\n### Asynchronous API Client\n\nFor high-performance applications, use the async client:\n\n```python\nfrom lastcron import AsyncAPIClient\nimport asyncio\n\nasync def main():\n async with AsyncAPIClient(token=\"your_token\", base_url=\"http://localhost/api\") as client:\n # Trigger multiple flows concurrently\n tasks = [\n client.trigger_flow_by_name(1, \"flow_a\"),\n client.trigger_flow_by_name(1, \"flow_b\"),\n client.trigger_flow_by_name(1, \"flow_c\")\n ]\n results = await asyncio.gather(*tasks)\n\nasyncio.run(main())\n```\n\n### Type-Safe Data Classes\n\nUse strongly-typed dataclasses for better IDE support:\n\n```python\nfrom lastcron import flow, get_run_logger, FlowRun, Block, BlockType\n\n@flow\ndef typed_flow(**params):\n logger = get_run_logger()\n \n # Trigger a flow and get a typed response\n run: FlowRun = my_other_flow.submit(parameters={'key': 'value'})\n \n if run:\n logger.info(f\"Run ID: {run.id}\")\n logger.info(f\"State: {run.state.value}\")\n \n # Get a block with type information\n config: Block = get_block('config')\n if config and config.type == BlockType.JSON:\n logger.info(\"Got JSON configuration\")\n```\n\n## \ud83d\udee0\ufe0f Development\n\n### Setting Up Development Environment\n\n```bash\n# Clone the repository\ngit clone https://github.com/allanbru/lastcron-sdk.git\ncd lastcron-sdk\n\n# Create virtual environment\npython -m venv venv\nsource venv/bin/activate # On Windows: venv\\Scripts\\activate\n\n# Install in development mode\npip install -e \".[dev]\"\n```\n\n### Running Tests\n\n```bash\n# Run all tests\npytest\n\n# Run with coverage\npytest --cov=lastcron --cov-report=html\n\n# Run specific test file\npytest tests/test_flow.py\n```\n\n### Code Quality\n\n```bash\n# Format code\nblack lastcron/\n\n# Lint code\nruff check lastcron/\n\n# Type check\nmypy lastcron/\n```\n\n## \ud83d\udcd6 Documentation\n\nFor detailed documentation, see:\n\n- [SDK Documentation](lastcron/README.md) - Comprehensive SDK guide\n- [Contributing Guide](CONTRIBUTING.md) - How to contribute\n- [Examples](examples/) - Example flows and use cases\n\n## \ud83e\udd1d Contributing\n\nWe welcome contributions! Please see our [Contributing Guide](CONTRIBUTING.md) for details on:\n\n- Setting up your development environment\n- Code style and standards\n- Testing requirements\n- Pull request process\n\n## \ud83d\udcc4 License\n\nThis project is licensed under the **Elastic License 2.0** - see the [LICENSE](LICENSE) file for details.\n\nThe Elastic License 2.0 is a source-available license that allows you to:\n- Use the software freely\n- Modify and distribute the software\n- Use it in commercial applications\n\nWith the limitation that you cannot provide the software as a managed service.\n\n## \ud83d\ude4f Acknowledgments\n\nBuilt with \u2764\ufe0f by AllanBR Creations\n\n## \ud83d\udcde Support\n\n- **Issues**: [GitHub Issues](https://github.com/allanbru/lastcron-sdk/issues)\n- **Discussions**: [GitHub Discussions](https://github.com/allanbru/lastcron-sdk/discussions)\n\n## \ud83d\uddfa\ufe0f Roadmap\n\n- [ ] Enhanced error handling and retry mechanisms\n- [ ] Flow dependencies and DAG support\n- [ ] Real-time flow monitoring\n- [ ] Flow templates and reusable components\n- [ ] Integration with popular data tools\n- [ ] Web UI for flow visualization\n\n---\n\n**Note**: This SDK is designed to work with the LastCron orchestration platform. You'll need a running LastCron instance to execute flows.\n\n",
"bugtrack_url": null,
"license": "Elastic-2.0",
"summary": "Python SDK for building and orchestrating workflows with LastCron",
"version": "0.1.0",
"project_urls": {
"Changelog": "https://github.com/allanbru/lastcron-sdk/releases",
"Documentation": "https://github.com/allanbru/lastcron-sdk#readme",
"Homepage": "https://github.com/allanbru/lastcron-sdk",
"Issues": "https://github.com/allanbru/lastcron-sdk/issues",
"Repository": "https://github.com/allanbru/lastcron-sdk"
},
"split_keywords": [
"orchestration",
" workflow",
" cron",
" task-management",
" automation",
" scheduling",
" data-pipeline",
" etl"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "b0172d66dad378e35f11a7ad00bb5e6531b1709d4444f36e715ae3b1dd36f3a5",
"md5": "2d949975d85ab3a648d674565d043dee",
"sha256": "85725878aea446a86cc9d9c644ba24b46c2c7201959b327368cbdbf37995436a"
},
"downloads": -1,
"filename": "lastcron-0.1.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "2d949975d85ab3a648d674565d043dee",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 24451,
"upload_time": "2025-11-03T22:30:27",
"upload_time_iso_8601": "2025-11-03T22:30:27.828296Z",
"url": "https://files.pythonhosted.org/packages/b0/17/2d66dad378e35f11a7ad00bb5e6531b1709d4444f36e715ae3b1dd36f3a5/lastcron-0.1.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "a3931494b5fa76607ac43f58dd7d106a4fbc6064ec8f813882b9c52c3985f7f3",
"md5": "9bc838433914edd31eb6c15df73d5268",
"sha256": "9bf5e0c8f77cfc16a7724c28e7fc9c232e359ca2998f3bfccc891f8bd7e60be1"
},
"downloads": -1,
"filename": "lastcron-0.1.0.tar.gz",
"has_sig": false,
"md5_digest": "9bc838433914edd31eb6c15df73d5268",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 30589,
"upload_time": "2025-11-03T22:30:29",
"upload_time_iso_8601": "2025-11-03T22:30:29.155024Z",
"url": "https://files.pythonhosted.org/packages/a3/93/1494b5fa76607ac43f58dd7d106a4fbc6064ec8f813882b9c52c3985f7f3/lastcron-0.1.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-11-03 22:30:29",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "allanbru",
"github_project": "lastcron-sdk",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "lastcron"
}
JSON
