Refactor async operations from Celery to Dagster background tasks

.gitignore

@@ -93,11 +93,6 @@ ipython_config.py
 #   install all needed dependencies.
 Dockerfile
 
-# celery beat schedule file
-celerybeat-schedule.db
-celerybeat-schedule
-celerybeat.pid
-
 # SageMath parsed files
 *.sage.py
 

pytest.ini

@@ -5,7 +5,6 @@ python_classes = Test*
 python_functions = test_*
 
 markers =
-    celery: marks tests that require celery
     integration: marks integration tests
     unit: marks unit tests
     asyncio: marks tests that use asyncio

robosystems_client/api/agent/auto_select_agent.py

@@ -134,7 +134,7 @@ def sync_detailed(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -228,7 +228,7 @@ def sync(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -317,7 +317,7 @@ async def asyncio_detailed(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -409,7 +409,7 @@ async def asyncio(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.

robosystems_client/api/agent/execute_specific_agent.py

@@ -124,7 +124,7 @@ def sync_detailed(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -179,7 +179,7 @@ def sync(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -229,7 +229,7 @@ async def asyncio_detailed(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.
@@ -282,7 +282,7 @@ async def asyncio(
   **Execution Strategies (automatic):**
   - Fast operations (<5s): Immediate synchronous response
   - Medium operations (5-30s): SSE streaming with progress updates
-  - Long operations (>30s): Async Celery worker with operation tracking
+  - Long operations (>30s): Background queue with operation tracking
 
   **Response Mode Override:**
   Use query parameter `?mode=sync|async` to override automatic strategy selection.

robosystems_client/api/files/update_file.py

@@ -103,7 +103,7 @@ def sync_detailed(
   **What Happens (status='uploaded'):**
   1. File validated in S3
   2. Row count calculated
-  3. DuckDB staging triggered immediately (Celery task)
+  3. DuckDB staging triggered immediately (background task)
   4. If ingest_to_graph=true, graph ingestion queued
   5. File queryable in DuckDB within seconds
 
@@ -165,7 +165,7 @@ def sync(
   **What Happens (status='uploaded'):**
   1. File validated in S3
   2. Row count calculated
-  3. DuckDB staging triggered immediately (Celery task)
+  3. DuckDB staging triggered immediately (background task)
   4. If ingest_to_graph=true, graph ingestion queued
   5. File queryable in DuckDB within seconds
 
@@ -222,7 +222,7 @@ async def asyncio_detailed(
   **What Happens (status='uploaded'):**
   1. File validated in S3
   2. Row count calculated
-  3. DuckDB staging triggered immediately (Celery task)
+  3. DuckDB staging triggered immediately (background task)
   4. If ingest_to_graph=true, graph ingestion queued
   5. File queryable in DuckDB within seconds
 
@@ -282,7 +282,7 @@ async def asyncio(
   **What Happens (status='uploaded'):**
   1. File validated in S3
   2. Row count calculated
-  3. DuckDB staging triggered immediately (Celery task)
+  3. DuckDB staging triggered immediately (background task)
   4. If ingest_to_graph=true, graph ingestion queued
   5. File queryable in DuckDB within seconds
 

robosystems_client/extensions/README.md

@@ -1,6 +1,6 @@
 # RoboSystems Python Client Extensions
 
-🚀 **Production-Ready Extensions** for the RoboSystems Financial Knowledge Graph API
+**Production-Ready Extensions** for the RoboSystems Financial Knowledge Graph API
 
 [![Python 3.7+](https://img.shields.io/badge/python-3.7+-blue.svg)](https://www.python.org/downloads/)
 [![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
@@ -17,7 +17,7 @@ The RoboSystems Python Client Extensions provide enhanced functionality for the
 - **Caching** with TTL and LRU eviction
 - **Full Async/Await Support** throughout
 
-## 🚀 Quick Start
+## Quick Start
 
 ### Installation
 
@@ -84,7 +84,7 @@ async def main():
 asyncio.run(main())
 ```
 
-## 🔐 Authentication
+## Authentication
 
 ### API Key Authentication (Recommended)
 
@@ -139,7 +139,7 @@ dev_ext = create_extensions(
 )
 ```
 
-## 🛠 Advanced Features
+## Advanced Features
 
 ### Query Builder
 
@@ -182,7 +182,7 @@ print(f"Complexity: {cost['complexity_category']} (score: {cost['complexity_scor
 
 # Get optimization recommendations  
 for rec in cost['recommendations']:
-    print(f"💡 {rec}")
+    print(f"Tip: {rec}")
 ```
 
 ### Result Processing
@@ -271,7 +271,7 @@ client.connect("operation_id")
 client.close()
 ```
 
-## 📊 Examples
+## Examples
 
 ### Financial Data Analysis
 
@@ -332,7 +332,7 @@ if final_batch:
     process_transaction_batch(final_batch)
     total_processed += len(final_batch)
 
-print(f"✅ Processed {total_processed:,} transactions total")
+print(f"Processed {total_processed:,} transactions total")
 ```
 
 ### Error Handling
@@ -361,7 +361,7 @@ except Exception as e:
     print(f"Query failed: {e}")
 ```
 
-## ⚡ Performance Optimization
+## Performance Optimization
 
 ### Connection Pooling
 
@@ -391,21 +391,21 @@ query = "MATCH (c:Company) WHERE c.revenue > 1000000 RETURN c"
 # Check syntax
 validation = validate_cypher_query(query)
 if not validation['valid']:
-    print("❌ Query has syntax errors:", validation['issues'])
+    print("Query has syntax errors:", validation['issues'])
 
 # Estimate cost
 cost = estimate_query_cost(query)
-print(f"📊 Query complexity: {cost['complexity_category']}")
+print(f"Query complexity: {cost['complexity_category']}")
 
 # Follow recommendations
 for rec in cost['recommendations']:
-    print(f"💡 Optimization tip: {rec}")
+    print(f"Optimization tip: {rec}")
 
 # Execute only if reasonable complexity
 if cost['complexity_category'] in ['low', 'medium']:
     result = extensions.execute_query("graph_id", query)
 else:
-    print("⚠️ Query may be too expensive - consider optimization")
+    print("Query may be too expensive - consider optimization")
 ```
 
 ### Caching Strategy
@@ -425,7 +425,7 @@ else:
     cache = results_cache
 ```
 
-## 🧪 Testing
+## Testing
 
 Run the test suite:
 
@@ -457,7 +457,7 @@ def test_query_execution():
         assert result["data"] == [{"count": 100}]
 ```
 
-## 🔧 Configuration
+## Configuration
 
 ### Environment Variables
 
@@ -492,7 +492,7 @@ config = RoboSystemsExtensionConfig(
 extensions = RoboSystemsExtensions(config)
 ```
 
-## 📚 API Reference
+## API Reference
 
 ### Core Classes
 
@@ -528,7 +528,7 @@ extensions = RoboSystemsExtensions(config)
 - **`format_duration(milliseconds)`** - Human-readable time formatting
 - **`create_extensions(method, **kwargs)`** - Extensions factory
 
-## 🐛 Troubleshooting
+## Troubleshooting
 
 ### Common Issues
 
@@ -545,9 +545,9 @@ pip install pandas # For DataFrame conversion (optional)
 extensions = AuthenticatedExtensions("your-api-key")
 try:
     result = extensions.execute_query("graph_id", "MATCH (n) RETURN count(n) LIMIT 1")
-    print("✅ Authentication successful")
+    print("Authentication successful")
 except Exception as e:
-    print(f"❌ Auth failed: {e}")
+    print(f"Auth failed: {e}")
 ```
 
 **Connection Issues**
@@ -585,19 +585,15 @@ logging.basicConfig(level=logging.DEBUG)
 extensions = AuthenticatedExtensions("your-key")
 ```
 
-## 📄 License
-
-MIT License - see [LICENSE](LICENSE) file for details.
-
-## 🤝 Contributing
+## Contributing
 
 1. Fork the repository
 2. Create a feature branch
 3. Add tests for new functionality  
 4. Run the test suite: `python run_tests.py`
 5. Submit a pull request
 
-## 📞 Support
+## Support
 
 - **API Reference**: [api.robosystems.ai](https://api.robosystems.ai)
 - **Issues**: [GitHub Issues](https://github.com/RoboFinSystems/robosystems-python-client/issues)

robosystems_client/extensions/agent_client.py

@@ -177,7 +177,7 @@ def execute_query(
               else datetime.now().isoformat(),
             )
 
-        # Check if this is a queued response (async Celery execution)
+        # Check if this is a queued response (async background task execution)
         is_queued = False
         queued_response = None