You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
88 lines
2.6 KiB
88 lines
2.6 KiB
import {Connection, ConnectionNotReadyError} from './Connection'
|
|
import {Client} from 'pg'
|
|
import {Inject} from '../../di'
|
|
import {QueryResult} from '../types'
|
|
import {Awaitable, collect} from '../../util'
|
|
import {SQLDialect} from '../dialect/SQLDialect'
|
|
import {PostgreSQLDialect} from '../dialect/PostgreSQLDialect'
|
|
import {Logging} from '../../service/Logging'
|
|
import {Schema} from '../schema/Schema'
|
|
import {PostgresSchema} from '../schema/PostgresSchema'
|
|
|
|
/**
|
|
* Type interface representing the config for a PostgreSQL connection.
|
|
*/
|
|
export interface PostgresConnectionConfig {
|
|
user: string,
|
|
host: string,
|
|
database: string,
|
|
password?: string,
|
|
port?: string,
|
|
}
|
|
|
|
/**
|
|
* An implementation of a database Connection for dealing with PostgreSQL servers.
|
|
*/
|
|
export class PostgresConnection extends Connection {
|
|
@Inject()
|
|
protected readonly logging!: Logging
|
|
|
|
/** The `pg` database client. */
|
|
protected client?: Client
|
|
|
|
public dialect(): SQLDialect {
|
|
return <PostgreSQLDialect> this.app().make(PostgreSQLDialect)
|
|
}
|
|
|
|
public async init(): Promise<void> {
|
|
this.logging.debug(`Initializing PostgreSQL connection ${this.name}...`)
|
|
this.client = new Client(this.config)
|
|
await this.client.connect()
|
|
}
|
|
|
|
public async close(): Promise<void> {
|
|
this.logging.debug(`Closing PostgreSQL connection ${this.name}...`)
|
|
if ( this.client ) {
|
|
await this.client.end()
|
|
}
|
|
}
|
|
|
|
public async query(query: string): Promise<QueryResult> {
|
|
if ( !this.client ) {
|
|
throw new ConnectionNotReadyError(this.name, { config: JSON.stringify(this.config) })
|
|
}
|
|
this.logging.verbose(`Executing query in connection ${this.name}: \n${query.split('\n').map(x => ' ' + x)
|
|
.join('\n')}`)
|
|
|
|
try {
|
|
const result = await this.client.query(query)
|
|
await this.queryExecuted(query)
|
|
|
|
return {
|
|
rows: collect(result.rows),
|
|
rowCount: result.rowCount,
|
|
}
|
|
} catch (e) {
|
|
throw this.app().errorWrapContext(e, {
|
|
query,
|
|
connection: this.name,
|
|
})
|
|
}
|
|
}
|
|
|
|
public async asTransaction<T>(closure: () => Awaitable<T>): Promise<T> {
|
|
if ( !this.client ) {
|
|
throw new ConnectionNotReadyError(this.name, { config: JSON.stringify(this.config) })
|
|
}
|
|
|
|
await this.client.query('BEGIN')
|
|
const result = await closure()
|
|
await this.client.query('COMMIT')
|
|
return result
|
|
}
|
|
|
|
public schema(name?: string): Schema {
|
|
return new PostgresSchema(this, name)
|
|
}
|
|
}
|