0
# RDS Operations
1

2
Connect to and manage Amazon RDS instances for relational database operations within Dagster pipelines.
3

4
## Capabilities
5

6
### RDS Resource
7

8
```python { .api }
9
class RDSResource(ResourceWithBoto3Configuration):
10
    """
11
    Resource for interacting with Amazon RDS.
12
    """
13
    
14
    def get_connection(self, db_instance_identifier: str):
15
        """
16
        Get connection to RDS database instance.
17
        
18
        Parameters:
19
            db_instance_identifier: RDS instance identifier
20
            
21
        Returns:
22
            Connection: Database connection object
23
        """
24
    
25
    def describe_db_instances(self) -> List[Dict]:
26
        """
27
        List available RDS instances.
28
        
29
        Returns:
30
            List[Dict]: RDS instance descriptions
31
        """
32
    
33
    def create_db_snapshot(
34
        self, 
35
        db_instance_identifier: str,
36
        snapshot_identifier: str
37
    ) -> Dict:
38
        """
39
        Create RDS database snapshot.
40
        
41
        Parameters:
42
            db_instance_identifier: Source RDS instance
43
            snapshot_identifier: Snapshot name
44
            
45
        Returns:
46
            Dict: Snapshot creation response
47
        """
48
```
49

50
## Usage Examples
51

52
```python
53
from dagster import op, job, Definitions
54
from dagster_aws.rds import RDSResource
55

56
@op(required_resource_keys={"rds"})
57
def backup_database(context):
58
    rds = context.resources.rds
59
    
60
    # Create database snapshot
61
    snapshot_id = f"backup-{context.run_id}"
62
    
63
    response = rds.create_db_snapshot(
64
        db_instance_identifier="my-prod-database",
65
        snapshot_identifier=snapshot_id
66
    )
67
    
68
    context.log.info(f"Created snapshot: {snapshot_id}")
69
    return response
70

71
@job(
72
    resource_defs={
73
        "rds": RDSResource(region_name="us-west-2")
74
    }
75
)
76
def database_backup_job():
77
    backup_database()
78

79
defs = Definitions(jobs=[database_backup_job])
80
```