athena_rs 3.22.1

Hyper performant polyglot Database driver
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

//! Shared request context helpers for schema catalog route handlers.
//!
//! This module centralizes request-scoped schema-client resolution so endpoint
//! handlers stay focused on input normalization and response flow.

use actix_web::{HttpRequest, HttpResponse};
use sqlx::{Pool, Postgres};

use crate::AppState;
use crate::api::client_context::required_client_name;
use crate::api::response::bad_request;
use crate::athena::resolver::{
    AthenaClientResolveError, AthenaResolvedQueryBackend, resolve_query_backend,
};
use crate::drivers::cloudflare_d1::client::D1ConnectionInfo;

use super::context_pool_resolver::require_schema_client_pool;

#[derive(Debug, Clone)]
pub(crate) enum SchemaBackend {
    Postgres { pool: Pool<Postgres> },
    D1 { connection_info: D1ConnectionInfo },
}

/// Resolves and returns the schema client pool for request-scoped catalog calls.
pub(super) async fn require_schema_pool(
    req: &HttpRequest,
    app_state: &AppState,
) -> Result<Pool<Postgres>, HttpResponse> {
    let (_client_name, pool): (String, Pool<Postgres>) =
        match require_schema_client_pool(req, app_state).await {
            Ok(resolved) => resolved,
            Err(resp) => return Err(resp),
        };
    Ok(pool)
}

/// Resolves and returns the schema backend for request-scoped catalog calls.
pub(super) async fn require_schema_backend(
    req: &HttpRequest,
    app_state: &AppState,
) -> Result<SchemaBackend, HttpResponse> {
    let client_name: String = match required_client_name(req) {
        Ok(client_name) => client_name,
        Err(resp) => return Err(resp),
    };

    if let Err(resp) =
        super::context_auth::authorize_schema_read(req, app_state, Some(client_name.as_str())).await
    {
        return Err(resp);
    }

    match resolve_query_backend(app_state, &client_name).await {
        Ok(Some(AthenaResolvedQueryBackend::D1 {
            connection_info, ..
        })) => {
            return Ok(SchemaBackend::D1 { connection_info });
        }
        Ok(Some(AthenaResolvedQueryBackend::Scylla { .. })) => {
            return Err(bad_request(
                "Unsupported schema backend",
                "Schema catalog routes do not support Scylla clients.",
            ));
        }
        Ok(Some(AthenaResolvedQueryBackend::Postgres(_))) | Ok(None) => {}
        Err(err) => {
            return Err(match err {
                AthenaClientResolveError::Inactive { client_name } => bad_request(
                    format!("Client '{}' is inactive", client_name),
                    "Schema client is configured but inactive",
                ),
                AthenaClientResolveError::Frozen { client_name } => bad_request(
                    format!("Client '{}' is frozen", client_name),
                    "Schema client is configured but frozen",
                ),
                AthenaClientResolveError::InvalidMetadata {
                    client_name,
                    message,
                } => bad_request(
                    format!("Client '{}' has invalid metadata", client_name),
                    message,
                ),
                AthenaClientResolveError::Lookup {
                    client_name,
                    message,
                } => bad_request(format!("Client '{}' lookup failed", client_name), message),
            });
        }
    }

    let pool = match require_schema_client_pool(req, app_state).await {
        Ok((_, pool)) => pool,
        Err(resp) => return Err(resp),
    };

    Ok(SchemaBackend::Postgres { pool })
}