Struct DatabaseSource 

pub struct DatabaseSource { /* private fields */ }

Adapter: PostgreSQL database as a pipeline data source.

Wraps a sqlx::PgPool and implements both DataSourcePort (for direct querying) and ScrapingService (for DAG pipeline integration).

Example

use stygian_graph::adapters::database::DatabaseSource;

let db = DatabaseSource::new("postgres://localhost/testdb").await.unwrap();
println!("Connected to: {}", db.source_name());

Implementations

impl DatabaseSource

pub async fn new(database_url: &str) -> Result<Self>

Connect to a PostgreSQL database.

Arguments

• database_url - PostgreSQL connection string

Example

use stygian_graph::adapters::database::DatabaseSource;

let db = DatabaseSource::new("postgres://localhost/mydb").await.unwrap();

pub fn from_pool(pool: PgPool) -> Self

Create from an existing pool (useful for testing or shared connections).

Example

use stygian_graph::adapters::database::DatabaseSource;
use sqlx::PgPool;

let db = DatabaseSource::from_pool(pool);

pub fn source_name(&self) -> &str

Return the name of this source for display/logging.

Trait Implementations

impl DataSourcePort for DatabaseSource

fn query<'life0, 'async_trait>( &'life0 self, params: QueryParams, ) -> Pin<Box<dyn Future<Output = Result<Vec<Value>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Execute a query and return results as JSON rows.

fn healthcheck<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Check that the underlying connection is alive.

fn source_name(&self) -> &str

Human-readable name of this data source (e.g. "postgres", "sqlite").

impl ScrapingService for DatabaseSource

fn execute<'life0, 'async_trait>( &'life0 self, input: ServiceInput, ) -> Pin<Box<dyn Future<Output = Result<ServiceOutput>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Execute a database query from pipeline parameters.

Expected params:

{ "query": "SELECT ...", "parameters": [...], "limit": 100 }

Example

let input = ServiceInput {
    url: String::new(),
    params: json!({"query": "SELECT 1 AS n", "parameters": [], "limit": 10}),
};
let result = db.execute(input).await.unwrap();

fn name(&self) -> &'static str

Service name for identification in logs and metrics