# fbapi - File-Based API Communication Library
A Python library for file-based API communication between processes using the filesystem as a communication medium.
## Features
**Event-Driven Architecture**
- Event-driven file monitoring using `watchdog` library
- Automatic fallback to polling when `watchdog` unavailable
- Real-time command processing with minimal latency
**Security**
- Path traversal attack prevention
- File permission validation
- Content security scanning
- Configurable file size limits
**Configuration**
- YAML and JSON configuration file support
- Environment variable overrides
- Runtime configuration updates
- Default configuration templates
**Testing**
- Test suite with unit, integration, and performance tests
- Mock filesystem testing
- Cross-platform compatibility
**Developer Tools**
- Command-line interface for testing and debugging
- Error handling with typed exceptions
- Logging and monitoring
- Type hints and modern Python practices
## Quick Start
### Installation
```bash
pip install fbapi
```
For development with all features:
```bash
pip install fbapi[dev]
```
### Basic Usage
#### Server Setup
```python
from fbapi import FileBasedAPIServer, EventSystem
# Create event system and register handlers
event_system = EventSystem()
def hello_handler(command_data):
params = command_data.get('params', [])
name = next((p['value'] for p in params if p['name'] == 'name'), 'World')
return {
'name': 'greeting',
'type': 'string',
'value': f'Hello, {name}!'
}
event_system.on('hello', hello_handler)
# Start server
server = FileBasedAPIServer(
command_dir="./commands",
response_dir="./responses",
event_system=event_system
)
server.start()
```
#### Client Usage
```python
from fbapi import FileBasedAPIClient
# Create client
client = FileBasedAPIClient(
command_dir="./commands",
response_dir="./responses"
)
# Define response handler
def handle_response(response_data):
if response_data['status'] == 'success':
result = response_data['response'][0]['value']
print(f"Received: {result}")
# Send command
client.call_command('hello', handle_response, name='Alice')
# Wait for completion
client.wait_for_completion()
```
### Configuration
Create a configuration file:
```bash
fbapi create-config fbapi_config.json
```
Use configuration in your application:
```python
from fbapi.config import load_config
config = load_config('fbapi_config.json')
client = FileBasedAPIClient(
command_dir=config.get('directories.command_dir'),
response_dir=config.get('directories.response_dir'),
timeout_seconds=config.get('client.timeout_seconds')
)
```
### Environment Variables
Override configuration with environment variables:
```bash
export FBAPI_CLIENT_TIMEOUT=30.0
export FBAPI_CLIENT_MONITORING=event
export FBAPI_LOG_LEVEL=DEBUG
```
## Command Line Interface
### Test Client-Server Communication
```bash
# Start test server (in one terminal)
fbapi test-server --command echo
# Send test command (in another terminal)
fbapi test-client --command echo --message "Hello fbapi!"
```
### Monitor Directory Activity
```bash
fbapi monitor --directory ./commands --strategy event
```
### Validate JSON Files
```bash
fbapi validate command1.json command2.json --schema-type request
```
## Advanced Features
### Security Configuration
```python
from fbapi.security import SecurityValidator
security_validator = SecurityValidator(
allowed_base_paths=["/safe/directory"],
max_file_size=1024*1024, # 1MB limit
allowed_extensions=[".json"]
)
client = FileBasedAPIClient(
command_dir="./commands",
response_dir="./responses",
security_validator=security_validator
)
```
### Custom Monitoring Strategy
```python
# Force polling mode (useful for network filesystems)
client = FileBasedAPIClient(
command_dir="./commands",
response_dir="./responses",
monitoring_strategy="polling",
polling_interval=0.5 # Poll every 500ms
)
# Force event-driven mode (requires watchdog)
client = FileBasedAPIClient(
command_dir="./commands",
response_dir="./responses",
monitoring_strategy="event"
)
```
### Error Handling
```python
from fbapi.exceptions import ValidationError, SecurityError, TimeoutError
try:
client.call_command('test', callback, param="value")
except ValidationError as e:
print(f"Validation failed: {e}")
except SecurityError as e:
print(f"Security violation: {e}")
except TimeoutError as e:
print(f"Operation timed out: {e}")
```
### Middleware and Event Processing
```python
event_system = EventSystem()
# Add middleware for logging
def logging_middleware(event_name, *args, **kwargs):
print(f"Processing event: {event_name}")
return args, kwargs
event_system.add_middleware(logging_middleware)
# Register multiple handlers
event_system.on('process_data', data_processor)
event_system.on('send_email', email_sender)
event_system.on('log_event', event_logger)
```
## Architecture
### Communication Flow
```
Client Filesystem Server
│ │ │
├─ Create command.json ────┤ │
│ ├─ Monitor for files ──────┤
│ │ ├─ Process command
│ │ ├─ Generate response
│ ├─ Create response.json ────┤
├─ Monitor for response ───┤ │
├─ Process response │ │
```
### File Formats
#### Command Format
```json
{
"command": "hello",
"request_id": "uuid-string",
"params": [
{
"name": "username",
"type": "string",
"value": "alice"
}
],
"response_file": "cmd_uuid.json"
}
```
#### Response Format
```json
{
"request_id": "uuid-string",
"status": "success",
"response": [
{
"name": "result",
"type": "string",
"value": "Hello, alice!"
}
]
}
```
#### Error Response Format
```json
{
"request_id": "uuid-string",
"status": "error",
"error": {
"code": 400,
"message": "Validation failed: missing required field"
}
}
```
## Performance
### Benchmarks
| Feature | Event-Driven | Polling (1s) | Polling (0.1s) |
|---------|-------------|--------------|----------------|
| Response Time | ~10ms | ~500ms | ~50ms |
| CPU Usage | Low | Very Low | Low |
| File System Load | Minimal | Very Low | Low |
### Optimization
1. Use event-driven monitoring when possible for best performance
2. Tune polling intervals based on your latency requirements
3. Implement connection pooling for high-frequency communication
4. Use appropriate file size limits to prevent resource exhaustion
5. Configure logging levels appropriately for production
## Development
### Running Tests
```bash
# Install development dependencies
pip install -e .[dev]
# Run all tests
pytest
# Run with coverage
pytest --cov=fbapi --cov-report=html
# Run only unit tests
pytest tests/ -m "not integration"
# Run integration tests
pytest tests/ -m integration
```
### Code Quality
```bash
# Format code
black fbapi/
isort fbapi/
# Lint code
flake8 fbapi/
mypy fbapi/
```
## Contributing
1. Fork the repository
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
3. Make your changes with tests
4. Ensure all tests pass (`pytest`)
5. Commit your changes (`git commit -m 'Add amazing feature'`)
6. Push to the branch (`git push origin feature/amazing-feature`)
7. Open a Pull Request
## License
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
## Support
- **Documentation**: [https://fbapi.readthedocs.io/](https://fbapi.readthedocs.io/)
- **Issues**: [https://github.com/your-username/fbapi/issues](https://github.com/your-username/fbapi/issues)
- **Discussions**: [https://github.com/your-username/fbapi/discussions](https://github.com/your-username/fbapi/discussions)
## Changelog
### v0.1.0 (2025-08-16)
- Initial release
- Event-driven and polling-based monitoring
- Security features
- Configuration management
- Command-line interface
- Test suite
- Documentation and examples
Raw data
{
"_id": null,
"home_page": null,
"name": "fbapi",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "api, file-based, ipc, communication, polling",
"author": "fbapi contributors",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/e8/42/667f514f02569c15a9e574dcfd47b13c111d2220e462724760a2061a4596/fbapi-0.5.11962.tar.gz",
"platform": null,
"description": "# fbapi - File-Based API Communication Library\n\nA Python library for file-based API communication between processes using the filesystem as a communication medium.\n\n## Features\n\n**Event-Driven Architecture**\n- Event-driven file monitoring using `watchdog` library\n- Automatic fallback to polling when `watchdog` unavailable\n- Real-time command processing with minimal latency\n\n**Security**\n- Path traversal attack prevention\n- File permission validation\n- Content security scanning\n- Configurable file size limits\n\n**Configuration**\n- YAML and JSON configuration file support\n- Environment variable overrides\n- Runtime configuration updates\n- Default configuration templates\n\n**Testing**\n- Test suite with unit, integration, and performance tests\n- Mock filesystem testing\n- Cross-platform compatibility\n\n**Developer Tools**\n- Command-line interface for testing and debugging\n- Error handling with typed exceptions\n- Logging and monitoring\n- Type hints and modern Python practices\n\n## Quick Start\n\n### Installation\n\n```bash\npip install fbapi\n```\n\nFor development with all features:\n```bash\npip install fbapi[dev]\n```\n\n### Basic Usage\n\n#### Server Setup\n\n```python\nfrom fbapi import FileBasedAPIServer, EventSystem\n\n# Create event system and register handlers\nevent_system = EventSystem()\n\ndef hello_handler(command_data):\n params = command_data.get('params', [])\n name = next((p['value'] for p in params if p['name'] == 'name'), 'World')\n \n return {\n 'name': 'greeting',\n 'type': 'string', \n 'value': f'Hello, {name}!'\n }\n\nevent_system.on('hello', hello_handler)\n\n# Start server\nserver = FileBasedAPIServer(\n command_dir=\"./commands\",\n response_dir=\"./responses\", \n event_system=event_system\n)\n\nserver.start()\n```\n\n#### Client Usage\n\n```python\nfrom fbapi import FileBasedAPIClient\n\n# Create client\nclient = FileBasedAPIClient(\n command_dir=\"./commands\",\n response_dir=\"./responses\"\n)\n\n# Define response handler\ndef handle_response(response_data):\n if response_data['status'] == 'success':\n result = response_data['response'][0]['value']\n print(f\"Received: {result}\")\n\n# Send command\nclient.call_command('hello', handle_response, name='Alice')\n\n# Wait for completion\nclient.wait_for_completion()\n```\n\n### Configuration\n\nCreate a configuration file:\n\n```bash\nfbapi create-config fbapi_config.json\n```\n\nUse configuration in your application:\n\n```python\nfrom fbapi.config import load_config\n\nconfig = load_config('fbapi_config.json')\n\nclient = FileBasedAPIClient(\n command_dir=config.get('directories.command_dir'),\n response_dir=config.get('directories.response_dir'),\n timeout_seconds=config.get('client.timeout_seconds')\n)\n```\n\n### Environment Variables\n\nOverride configuration with environment variables:\n\n```bash\nexport FBAPI_CLIENT_TIMEOUT=30.0\nexport FBAPI_CLIENT_MONITORING=event\nexport FBAPI_LOG_LEVEL=DEBUG\n```\n\n## Command Line Interface\n\n### Test Client-Server Communication\n\n```bash\n# Start test server (in one terminal)\nfbapi test-server --command echo\n\n# Send test command (in another terminal) \nfbapi test-client --command echo --message \"Hello fbapi!\"\n```\n\n### Monitor Directory Activity\n\n```bash\nfbapi monitor --directory ./commands --strategy event\n```\n\n### Validate JSON Files\n\n```bash\nfbapi validate command1.json command2.json --schema-type request\n```\n\n## Advanced Features\n\n### Security Configuration\n\n```python\nfrom fbapi.security import SecurityValidator\n\nsecurity_validator = SecurityValidator(\n allowed_base_paths=[\"/safe/directory\"],\n max_file_size=1024*1024, # 1MB limit\n allowed_extensions=[\".json\"]\n)\n\nclient = FileBasedAPIClient(\n command_dir=\"./commands\",\n response_dir=\"./responses\",\n security_validator=security_validator\n)\n```\n\n### Custom Monitoring Strategy\n\n```python\n# Force polling mode (useful for network filesystems)\nclient = FileBasedAPIClient(\n command_dir=\"./commands\",\n response_dir=\"./responses\", \n monitoring_strategy=\"polling\",\n polling_interval=0.5 # Poll every 500ms\n)\n\n# Force event-driven mode (requires watchdog)\nclient = FileBasedAPIClient(\n command_dir=\"./commands\",\n response_dir=\"./responses\",\n monitoring_strategy=\"event\" \n)\n```\n\n### Error Handling\n\n```python\nfrom fbapi.exceptions import ValidationError, SecurityError, TimeoutError\n\ntry:\n client.call_command('test', callback, param=\"value\")\nexcept ValidationError as e:\n print(f\"Validation failed: {e}\")\nexcept SecurityError as e:\n print(f\"Security violation: {e}\")\nexcept TimeoutError as e:\n print(f\"Operation timed out: {e}\")\n```\n\n### Middleware and Event Processing\n\n```python\nevent_system = EventSystem()\n\n# Add middleware for logging\ndef logging_middleware(event_name, *args, **kwargs):\n print(f\"Processing event: {event_name}\")\n return args, kwargs\n\nevent_system.add_middleware(logging_middleware)\n\n# Register multiple handlers\nevent_system.on('process_data', data_processor)\nevent_system.on('send_email', email_sender)\nevent_system.on('log_event', event_logger)\n```\n\n## Architecture\n\n### Communication Flow\n\n```\nClient Filesystem Server\n \u2502 \u2502 \u2502\n \u251c\u2500 Create command.json \u2500\u2500\u2500\u2500\u2524 \u2502\n \u2502 \u251c\u2500 Monitor for files \u2500\u2500\u2500\u2500\u2500\u2500\u2524\n \u2502 \u2502 \u251c\u2500 Process command\n \u2502 \u2502 \u251c\u2500 Generate response\n \u2502 \u251c\u2500 Create response.json \u2500\u2500\u2500\u2500\u2524\n \u251c\u2500 Monitor for response \u2500\u2500\u2500\u2524 \u2502\n \u251c\u2500 Process response \u2502 \u2502\n```\n\n### File Formats\n\n#### Command Format\n```json\n{\n \"command\": \"hello\",\n \"request_id\": \"uuid-string\",\n \"params\": [\n {\n \"name\": \"username\", \n \"type\": \"string\",\n \"value\": \"alice\"\n }\n ],\n \"response_file\": \"cmd_uuid.json\"\n}\n```\n\n#### Response Format\n```json\n{\n \"request_id\": \"uuid-string\",\n \"status\": \"success\",\n \"response\": [\n {\n \"name\": \"result\",\n \"type\": \"string\", \n \"value\": \"Hello, alice!\"\n }\n ]\n}\n```\n\n#### Error Response Format\n```json\n{\n \"request_id\": \"uuid-string\", \n \"status\": \"error\",\n \"error\": {\n \"code\": 400,\n \"message\": \"Validation failed: missing required field\"\n }\n}\n```\n\n## Performance\n\n### Benchmarks\n\n| Feature | Event-Driven | Polling (1s) | Polling (0.1s) |\n|---------|-------------|--------------|----------------|\n| Response Time | ~10ms | ~500ms | ~50ms |\n| CPU Usage | Low | Very Low | Low |\n| File System Load | Minimal | Very Low | Low |\n\n### Optimization\n\n1. Use event-driven monitoring when possible for best performance\n2. Tune polling intervals based on your latency requirements \n3. Implement connection pooling for high-frequency communication\n4. Use appropriate file size limits to prevent resource exhaustion\n5. Configure logging levels appropriately for production\n\n## Development\n\n### Running Tests\n\n```bash\n# Install development dependencies\npip install -e .[dev]\n\n# Run all tests\npytest\n\n# Run with coverage\npytest --cov=fbapi --cov-report=html\n\n# Run only unit tests\npytest tests/ -m \"not integration\"\n\n# Run integration tests\npytest tests/ -m integration\n```\n\n### Code Quality\n\n```bash\n# Format code\nblack fbapi/\nisort fbapi/\n\n# Lint code \nflake8 fbapi/\nmypy fbapi/\n```\n\n## Contributing\n\n1. Fork the repository\n2. Create a feature branch (`git checkout -b feature/amazing-feature`)\n3. Make your changes with tests\n4. Ensure all tests pass (`pytest`)\n5. Commit your changes (`git commit -m 'Add amazing feature'`)\n6. Push to the branch (`git push origin feature/amazing-feature`)\n7. Open a Pull Request\n\n## License\n\nThis project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.\n\n## Support\n\n- **Documentation**: [https://fbapi.readthedocs.io/](https://fbapi.readthedocs.io/)\n- **Issues**: [https://github.com/your-username/fbapi/issues](https://github.com/your-username/fbapi/issues)\n- **Discussions**: [https://github.com/your-username/fbapi/discussions](https://github.com/your-username/fbapi/discussions)\n\n## Changelog\n\n### v0.1.0 (2025-08-16)\n- Initial release\n- Event-driven and polling-based monitoring\n- Security features\n- Configuration management\n- Command-line interface\n- Test suite\n- Documentation and examples\n",
"bugtrack_url": null,
"license": null,
"summary": "A general-purpose file-based API communication library",
"version": "0.5.11962",
"project_urls": {
"Documentation": "https://fbapi.readthedocs.io/",
"Homepage": "https://github.com/your-username/fbapi",
"Issues": "https://github.com/your-username/fbapi/issues",
"Repository": "https://github.com/your-username/fbapi.git"
},
"split_keywords": [
"api",
" file-based",
" ipc",
" communication",
" polling"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "245803f0efd5ead8d03b145fd6a7fe2c91c9d8bc9c69dcffb276a3a942395608",
"md5": "d2b3d61a3da3c56c11b7c46815f8aeff",
"sha256": "32d7360ed912ee6183793301f23aea653c122b26816c45db6cdf5a8cfd4bfa8d"
},
"downloads": -1,
"filename": "fbapi-0.5.11962-py3-none-any.whl",
"has_sig": false,
"md5_digest": "d2b3d61a3da3c56c11b7c46815f8aeff",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 24370,
"upload_time": "2025-08-16T14:36:44",
"upload_time_iso_8601": "2025-08-16T14:36:44.482640Z",
"url": "https://files.pythonhosted.org/packages/24/58/03f0efd5ead8d03b145fd6a7fe2c91c9d8bc9c69dcffb276a3a942395608/fbapi-0.5.11962-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "e842667f514f02569c15a9e574dcfd47b13c111d2220e462724760a2061a4596",
"md5": "f480c7815d89bfb79da24eedf33edd4d",
"sha256": "3289fe3d05059548626599bdf76e77dea66168ee5f8f4e13b2e8ebfb0d8dd2e7"
},
"downloads": -1,
"filename": "fbapi-0.5.11962.tar.gz",
"has_sig": false,
"md5_digest": "f480c7815d89bfb79da24eedf33edd4d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 79079,
"upload_time": "2025-08-16T14:36:45",
"upload_time_iso_8601": "2025-08-16T14:36:45.895333Z",
"url": "https://files.pythonhosted.org/packages/e8/42/667f514f02569c15a9e574dcfd47b13c111d2220e462724760a2061a4596/fbapi-0.5.11962.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-16 14:36:45",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "your-username",
"github_project": "fbapi",
"github_not_found": true,
"lcname": "fbapi"
}
JSON
