diff --git a/.changeset/five-donkeys-brake.md b/.changeset/five-donkeys-brake.md
new file mode 100644
index 0000000000..c9db3112a9
--- /dev/null
+++ b/.changeset/five-donkeys-brake.md
@@ -0,0 +1,50 @@
+---
+'example-backend': minor
+'@backstage/backend-common': minor
+---
+
+Introduces `PluginConnectionDatabaseManager`, a backwards compatible database
+connection manager which allows developers to configure database connections on
+a per plugin basis.
+
+The `backend.database` config path allows you to set `prefix` to use an
+alternate prefix for automatically generated database names, the default is
+`backstage_plugin_`. Use `backend.database.plugin.<pluginId>` to set plugin
+specific database connection configuration, e.g.
+
+```yaml
+backend:
+  database:
+    client: 'pg',
+    prefix: 'custom_prefix_'
+    connection:
+      host: 'localhost'
+      user: 'foo'
+      password: 'bar'
+    plugin:
+      catalog:
+        connection:
+          database: 'database_name_overriden'
+      scaffolder:
+        client: 'sqlite3'
+        connection: ':inmemory'
+```
+
+Existing backstage installations can be migrated by swapping out the database
+manager under `packages/backend/src/index.ts` as shown below:
+
+```diff
+import {
+-  SingleConnectionDatabaseManager,
++  PluginConnectionDatabaseManager,
+} from '@backstage/backend-common';
+
+// ...
+
+function makeCreateEnv(config: Config) {
+  // ...
+-  const databaseManager = SingleConnectionDatabaseManager.fromConfig(config);
++  const databaseManager = PluginConnectionDatabaseManager.fromConfig(config);
+  // ...
+}
+```
diff --git a/docs/tutorials/configuring-plugin-databases.md b/docs/tutorials/configuring-plugin-databases.md
new file mode 100644
index 0000000000..b6281cb0ae
--- /dev/null
+++ b/docs/tutorials/configuring-plugin-databases.md
@@ -0,0 +1,107 @@
+---
+id: configuring-plugin-databases
+title: Configuring Plugin Specific Databases
+# prettier-ignore
+description: Guide on how to use predefined databases for each plugin.
+---
+
+There are occasions where it may be difficult to deploy Backstage with
+automatically created databases in production due to access control or other
+restrictions. For example, your infrastructure might be defined as code using
+tools such as Terraform or AWS CloudFormation where the name of each database is
+defined, created and assigned explicitly.
+
+`@backstage/backend-common` provides an alternate database manager which allows
+you to set the client and database connection on a per plugin basis. This means
+that you can do selectively run certain plugins in memory with `sqlite3`, set
+different connection config including the name of the database and more.
+
+There are two additional configuration options for this database manager:
+
+- **`backend.database.prefix`:** is used to override the default
+  `backstage_plugin_` prefix which is used to generate a database name when it
+  is not explicitly set for that plugin.
+- **`backend.database.plugin.<pluginId>`:** is used to define a `client` and
+  `connection` block for the plugin matching the `pluginId`, e.g. `catalog` is
+  the `pluginId` for the catalog plugin and any configuration defined under that
+  block is specific to that plugin.
+
+## Install Database Drivers
+
+If you intend to use both `postgres` and `sqlite3`, you need to make sure the
+appropriate database drivers are installed in your `backend` package.
+
+```shell
+cd packages/backend
+
+# install pg if you need postgres
+yarn add pg
+
+# install sqlite3 if you intend to set it as the client
+yarn add sqlite3
+```
+
+## Add Configuration
+
+To override the default prefix, `backstage_plugin_`, set
+`backend.database.prefix` as shown below. This will use databases such as
+`my_company_catalog` and `my_company_auth` instead of `backstage_plugin_catalog`
+and `backstage_plugin_auth`.
+
+```yaml
+backend:
+  database:
+    client: pg
+    prefix: my_company_
+    connection:
+      host: localhost
+      user: postgres
+      password: password
+    plugin:
+      code-coverage:
+        connection:
+          database: pg_code_coverage_set_by_user
+```
+
+In the example above, the `code-coverage` plugin will use the same connection
+configuration defined under `database.connection` and use
+`pg_code_coverage_set_by_user` instead of `my_company_code-coverage` which would
+be automatically generated if a plugin configuration wasn't explicitly set.
+
+## Integrate `PluginConnectionDatabaseManager` into `backend`
+
+The `SingleConnectionDatabaseManager` used by default should be replaced with
+the `PluginConnectionDatabaseManager` in your `packages/backend/src/index.ts`
+file. Import the manager and replace the `.fromConfig` call as shown below:
+
+```diff
+import {
+-  SingleConnectionDatabaseManager,
++  PluginConnectionDatabaseManager,
+} from '@backstage/backend-common';
+
+// ...
+
+function makeCreateEnv(config: Config) {
+  // ...
+-  const databaseManager = SingleConnectionDatabaseManager.fromConfig(config);
++  const databaseManager = PluginConnectionDatabaseManager.fromConfig(config);
+  // ...
+}
+```
+
+## Check Your Databases
+
+The `PluginConnectionDatabaseManager` preserves the behaviour of the
+`SingleConnectionDatabaseManager`. If the database does not exist, it will
+attempt to create it. You should ensure the databases that you configure exists
+and that the connection details have the appropriate permissions to work with
+each of the given databases if you are using this database manager to set the
+database name upfront. If each database needs its own connection username,
+password or host - you may set them under the plugin's `connection` block.
+
+`sqlite3` databases do not need to be created upfront as with the existing
+database manager.
+
+Your Backstage App can now use different database clients and configuration per
+plugin!
diff --git a/packages/backend-common/api-report.md b/packages/backend-common/api-report.md
index 068220cad3..2648e7af13 100644
--- a/packages/backend-common/api-report.md
+++ b/packages/backend-common/api-report.md
@@ -3,7 +3,6 @@
 > Do not edit this file. It is a report generated by [API Extractor](https://api-extractor.com/).
 
 ```ts
-
 import { AzureIntegration } from '@backstage/integration';
 import { BitbucketIntegration } from '@backstage/integration';
 import { Config } from '@backstage/config';
@@ -31,95 +30,120 @@ import { Writable } from 'stream';
 
 // @public (undocumented)
 export class AzureUrlReader implements UrlReader {
-    constructor(integration: AzureIntegration, deps: {
-        treeResponseFactory: ReadTreeResponseFactory;
-    });
-    // (undocumented)
-    static factory: ReaderFactory;
-    // (undocumented)
-    read(url: string): Promise<Buffer>;
-    // (undocumented)
-    readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
-    // (undocumented)
-    search(url: string, options?: SearchOptions): Promise<SearchResponse>;
-    // (undocumented)
-    toString(): string;
+  constructor(
+    integration: AzureIntegration,
+    deps: {
+      treeResponseFactory: ReadTreeResponseFactory;
+    },
+  );
+  // (undocumented)
+  static factory: ReaderFactory;
+  // (undocumented)
+  read(url: string): Promise<Buffer>;
+  // (undocumented)
+  readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
+  // (undocumented)
+  search(url: string, options?: SearchOptions): Promise<SearchResponse>;
+  // (undocumented)
+  toString(): string;
 }
 
 // @public
 export class BitbucketUrlReader implements UrlReader {
-    constructor(integration: BitbucketIntegration, deps: {
-        treeResponseFactory: ReadTreeResponseFactory;
-    });
-    // (undocumented)
-    static factory: ReaderFactory;
-    // (undocumented)
-    read(url: string): Promise<Buffer>;
-    // (undocumented)
-    readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
-    // (undocumented)
-    search(url: string, options?: SearchOptions): Promise<SearchResponse>;
-    // (undocumented)
-    toString(): string;
+  constructor(
+    integration: BitbucketIntegration,
+    deps: {
+      treeResponseFactory: ReadTreeResponseFactory;
+    },
+  );
+  // (undocumented)
+  static factory: ReaderFactory;
+  // (undocumented)
+  read(url: string): Promise<Buffer>;
+  // (undocumented)
+  readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
+  // (undocumented)
+  search(url: string, options?: SearchOptions): Promise<SearchResponse>;
+  // (undocumented)
+  toString(): string;
 }
 
 // @public
 export interface CacheClient {
-    delete(key: string): Promise<void>;
-    get(key: string): Promise<JsonValue | undefined>;
-    set(key: string, value: JsonValue, options?: CacheSetOptions): Promise<void>;
+  delete(key: string): Promise<void>;
+  get(key: string): Promise<JsonValue | undefined>;
+  set(key: string, value: JsonValue, options?: CacheSetOptions): Promise<void>;
 }
 
 // @public
 export class CacheManager {
-    forPlugin(pluginId: string): PluginCacheManager;
-    static fromConfig(config: Config, options?: CacheManagerOptions): CacheManager;
-    }
+  forPlugin(pluginId: string): PluginCacheManager;
+  static fromConfig(config: Config, options?: CacheManagerOptions): CacheManager;
+}
 
 // @public (undocumented)
 export const coloredFormat: winston.Logform.Format;
 
 // @public (undocumented)
 export interface ContainerRunner {
-    // (undocumented)
-    runContainer(opts: RunContainerOptions): Promise<void>;
+  // (undocumented)
+  runContainer(opts: RunContainerOptions): Promise<void>;
 }
 
 // @public @deprecated
 export const createDatabase: typeof createDatabaseClient;
 
 // @public
-export function createDatabaseClient(dbConfig: Config, overrides?: Partial<Knex.Config>): Knex<any, unknown[]>;
+export function createDatabaseClient(
+  dbConfig: Config,
+  overrides?: Partial<Knex.Config>,
+): Knex<any, unknown[]>;
 
 // @public (undocumented)
-export function createRootLogger(options?: winston.LoggerOptions, env?: NodeJS.ProcessEnv): winston.Logger;
+export function createRootLogger(
+  options?: winston.LoggerOptions,
+  env?: NodeJS.ProcessEnv,
+): winston.Logger;
 
 // @public
 export function createServiceBuilder(_module: NodeModule): ServiceBuilderImpl;
 
 // @public (undocumented)
-export function createStatusCheckRouter(options: StatusCheckRouterOptions): Promise<express.Router>;
+export function createStatusCheckRouter(
+  options: StatusCheckRouterOptions,
+): Promise<express.Router>;
 
 // @public (undocumented)
 export class DockerContainerRunner implements ContainerRunner {
-    constructor({ dockerClient }: {
-        dockerClient: Docker;
-    });
-    // (undocumented)
-    runContainer({ imageName, command, args, logStream, mountDirs, workingDir, envVars, }: RunContainerOptions): Promise<void>;
+  constructor({ dockerClient }: { dockerClient: Docker });
+  // (undocumented)
+  runContainer({
+    imageName,
+    command,
+    args,
+    logStream,
+    mountDirs,
+    workingDir,
+    envVars,
+  }: RunContainerOptions): Promise<void>;
 }
 
 // @public
-export function ensureDatabaseExists(dbConfig: Config, ...databases: Array<string>): Promise<void>;
+export function ensureDatabaseExists(
+  dbConfig: Config,
+  ...databases: Array<string>
+): Promise<void>;
 
 // @public
-export function errorHandler(options?: ErrorHandlerOptions): ErrorRequestHandler;
+export function errorHandler(
+  options?: ErrorHandlerOptions,
+): ErrorRequestHandler;
 
 // @public (undocumented)
 export type ErrorHandlerOptions = {
-    showStackTraces?: boolean;
-    logger?: Logger;
-    logClientErrors?: boolean;
+  showStackTraces?: boolean;
+  logger?: Logger;
+  logClientErrors?: boolean;
 };
 
 // @public (undocumented)
@@ -130,120 +154,141 @@ export function getVoidLogger(): winston.Logger;
 
 // @public (undocumented)
 export class Git {
-    // (undocumented)
-    add({ dir, filepath, }: {
-        dir: string;
-        filepath: string;
-    }): Promise<void>;
-    // (undocumented)
-    addRemote({ dir, url, remote, }: {
-        dir: string;
-        remote: string;
-        url: string;
-    }): Promise<void>;
-    // (undocumented)
-    clone({ url, dir, ref, }: {
-        url: string;
-        dir: string;
-        ref?: string;
-    }): Promise<void>;
-    // (undocumented)
-    commit({ dir, message, author, committer, }: {
-        dir: string;
-        message: string;
-        author: {
-            name: string;
-            email: string;
-        };
-        committer: {
-            name: string;
-            email: string;
-        };
-    }): Promise<string>;
-    // (undocumented)
-    currentBranch({ dir, fullName, }: {
-        dir: string;
-        fullName?: boolean;
-    }): Promise<string | undefined>;
-    // (undocumented)
-    fetch({ dir, remote, }: {
-        dir: string;
-        remote?: string;
-    }): Promise<void>;
-    // (undocumented)
-    static fromAuth: ({ username, password, logger, }: {
-        username?: string | undefined;
-        password?: string | undefined;
-        logger?: Logger | undefined;
-    }) => Git;
-    // (undocumented)
-    init({ dir }: {
-        dir: string;
-    }): Promise<void>;
-    // (undocumented)
-    merge({ dir, theirs, ours, author, committer, }: {
-        dir: string;
-        theirs: string;
-        ours?: string;
-        author: {
-            name: string;
-            email: string;
-        };
-        committer: {
-            name: string;
-            email: string;
-        };
-    }): Promise<MergeResult>;
-    // (undocumented)
-    push({ dir, remote }: {
-        dir: string;
-        remote: string;
-    }): Promise<PushResult>;
-    // (undocumented)
-    readCommit({ dir, sha, }: {
-        dir: string;
-        sha: string;
-    }): Promise<ReadCommitResult>;
-    // (undocumented)
-    resolveRef({ dir, ref, }: {
-        dir: string;
-        ref: string;
-    }): Promise<string>;
+  // (undocumented)
+  add({ dir, filepath }: { dir: string; filepath: string }): Promise<void>;
+  // (undocumented)
+  addRemote({
+    dir,
+    url,
+    remote,
+  }: {
+    dir: string;
+    remote: string;
+    url: string;
+  }): Promise<void>;
+  // (undocumented)
+  clone({
+    url,
+    dir,
+    ref,
+  }: {
+    url: string;
+    dir: string;
+    ref?: string;
+  }): Promise<void>;
+  // (undocumented)
+  commit({
+    dir,
+    message,
+    author,
+    committer,
+  }: {
+    dir: string;
+    message: string;
+    author: {
+      name: string;
+      email: string;
+    };
+    committer: {
+      name: string;
+      email: string;
+    };
+  }): Promise<string>;
+  // (undocumented)
+  currentBranch({
+    dir,
+    fullName,
+  }: {
+    dir: string;
+    fullName?: boolean;
+  }): Promise<string | undefined>;
+  // (undocumented)
+  fetch({ dir, remote }: { dir: string; remote?: string }): Promise<void>;
+  // (undocumented)
+  static fromAuth: ({
+    username,
+    password,
+    logger,
+  }: {
+    username?: string | undefined;
+    password?: string | undefined;
+    logger?: Logger | undefined;
+  }) => Git;
+  // (undocumented)
+  init({ dir }: { dir: string }): Promise<void>;
+  // (undocumented)
+  merge({
+    dir,
+    theirs,
+    ours,
+    author,
+    committer,
+  }: {
+    dir: string;
+    theirs: string;
+    ours?: string;
+    author: {
+      name: string;
+      email: string;
+    };
+    committer: {
+      name: string;
+      email: string;
+    };
+  }): Promise<MergeResult>;
+  // (undocumented)
+  push({ dir, remote }: { dir: string; remote: string }): Promise<PushResult>;
+  // (undocumented)
+  readCommit({
+    dir,
+    sha,
+  }: {
+    dir: string;
+    sha: string;
+  }): Promise<ReadCommitResult>;
+  // (undocumented)
+  resolveRef({ dir, ref }: { dir: string; ref: string }): Promise<string>;
 }
 
 // @public
 export class GithubUrlReader implements UrlReader {
-    constructor(integration: GitHubIntegration, deps: {
-        treeResponseFactory: ReadTreeResponseFactory;
-        credentialsProvider: GithubCredentialsProvider;
-    });
-    // (undocumented)
-    static factory: ReaderFactory;
-    // (undocumented)
-    read(url: string): Promise<Buffer>;
-    // (undocumented)
-    readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
-    // (undocumented)
-    search(url: string, options?: SearchOptions): Promise<SearchResponse>;
-    // (undocumented)
-    toString(): string;
+  constructor(
+    integration: GitHubIntegration,
+    deps: {
+      treeResponseFactory: ReadTreeResponseFactory;
+      credentialsProvider: GithubCredentialsProvider;
+    },
+  );
+  // (undocumented)
+  static factory: ReaderFactory;
+  // (undocumented)
+  read(url: string): Promise<Buffer>;
+  // (undocumented)
+  readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
+  // (undocumented)
+  search(url: string, options?: SearchOptions): Promise<SearchResponse>;
+  // (undocumented)
+  toString(): string;
 }
 
 // @public (undocumented)
 export class GitlabUrlReader implements UrlReader {
-    constructor(integration: GitLabIntegration, deps: {
-        treeResponseFactory: ReadTreeResponseFactory;
-    });
-    // (undocumented)
-    static factory: ReaderFactory;
-    // (undocumented)
-    read(url: string): Promise<Buffer>;
-    // (undocumented)
-    readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
-    // (undocumented)
-    search(url: string, options?: SearchOptions): Promise<SearchResponse>;
-    // (undocumented)
-    toString(): string;
+  constructor(
+    integration: GitLabIntegration,
+    deps: {
+      treeResponseFactory: ReadTreeResponseFactory;
+    },
+  );
+  // (undocumented)
+  static factory: ReaderFactory;
+  // (undocumented)
+  read(url: string): Promise<Buffer>;
+  // (undocumented)
+  readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
+  // (undocumented)
+  search(url: string, options?: SearchOptions): Promise<SearchResponse>;
+  // (undocumented)
+  toString(): string;
 }
 
 // @public
@@ -254,32 +299,32 @@ export function notFoundHandler(): RequestHandler;
 
 // @public
 export type PluginCacheManager = {
-    getClient: (options?: ClientOptions) => CacheClient;
+  getClient: (options?: ClientOptions) => CacheClient;
 };
 
 // @public
 export interface PluginDatabaseManager {
-    getClient(): Promise<Knex>;
+  getClient(): Promise<Knex>;
 }
 
 // @public
 export type PluginEndpointDiscovery = {
-    getBaseUrl(pluginId: string): Promise<string>;
-    getExternalBaseUrl(pluginId: string): Promise<string>;
+  getBaseUrl(pluginId: string): Promise<string>;
+  getExternalBaseUrl(pluginId: string): Promise<string>;
 };
 
 // @public (undocumented)
 export type ReadTreeResponse = {
-    files(): Promise<ReadTreeResponseFile[]>;
-    archive(): Promise<NodeJS.ReadableStream>;
-    dir(options?: ReadTreeResponseDirOptions): Promise<string>;
-    etag: string;
+  files(): Promise<ReadTreeResponseFile[]>;
+  archive(): Promise<NodeJS.ReadableStream>;
+  dir(options?: ReadTreeResponseDirOptions): Promise<string>;
+  etag: string;
 };
 
 // @public
 export type ReadTreeResponseFile = {
-    path: string;
-    content(): Promise<Buffer>;
+  path: string;
+  content(): Promise<Buffer>;
 };
 
 // @public
@@ -290,37 +335,37 @@ export function resolvePackagePath(name: string, ...paths: string[]): string;
 
 // @public (undocumented)
 export type RunContainerOptions = {
-    imageName: string;
-    command?: string | string[];
-    args: string[];
-    logStream?: Writable;
-    mountDirs?: Record<string, string>;
-    workingDir?: string;
-    envVars?: Record<string, string>;
+  imageName: string;
+  command?: string | string[];
+  args: string[];
+  logStream?: Writable;
+  mountDirs?: Record<string, string>;
+  workingDir?: string;
+  envVars?: Record<string, string>;
 };
 
 // @public
 export type SearchResponse = {
-    files: SearchResponseFile[];
-    etag: string;
+  files: SearchResponseFile[];
+  etag: string;
 };
 
 // @public
 export type SearchResponseFile = {
-    url: string;
-    content(): Promise<Buffer>;
+  url: string;
+  content(): Promise<Buffer>;
 };
 
 // @public (undocumented)
 export type ServiceBuilder = {
-    loadConfig(config: ConfigReader): ServiceBuilder;
-    setPort(port: number): ServiceBuilder;
-    setHost(host: string): ServiceBuilder;
-    setLogger(logger: Logger): ServiceBuilder;
-    enableCors(options: cors.CorsOptions): ServiceBuilder;
-    setHttpsSettings(settings: HttpsSettings): ServiceBuilder;
-    addRouter(root: string, router: Router | RequestHandler): ServiceBuilder;
-    start(): Promise<Server>;
+  loadConfig(config: ConfigReader): ServiceBuilder;
+  setPort(port: number): ServiceBuilder;
+  setHost(host: string): ServiceBuilder;
+  setLogger(logger: Logger): ServiceBuilder;
+  enableCors(options: cors.CorsOptions): ServiceBuilder;
+  setHttpsSettings(settings: HttpsSettings): ServiceBuilder;
+  addRouter(root: string, router: Router | RequestHandler): ServiceBuilder;
+  start(): Promise<Server>;
 };
 
 // @public (undocumented)
@@ -328,52 +373,58 @@ export function setRootLogger(newLogger: winston.Logger): void;
 
 // @public
 export class SingleConnectionDatabaseManager {
-    forPlugin(pluginId: string): PluginDatabaseManager;
-    static fromConfig(config: Config): SingleConnectionDatabaseManager;
-    }
+  forPlugin(pluginId: string): PluginDatabaseManager;
+  static fromConfig(config: Config): SingleConnectionDatabaseManager;
+}
 
 // @public
 export class SingleHostDiscovery implements PluginEndpointDiscovery {
-    static fromConfig(config: Config, options?: {
-        basePath?: string;
-    }): SingleHostDiscovery;
-    // (undocumented)
-    getBaseUrl(pluginId: string): Promise<string>;
-    // (undocumented)
-    getExternalBaseUrl(pluginId: string): Promise<string>;
-    }
+  static fromConfig(
+    config: Config,
+    options?: {
+      basePath?: string;
+    },
+  ): SingleHostDiscovery;
+  // (undocumented)
+  getBaseUrl(pluginId: string): Promise<string>;
+  // (undocumented)
+  getExternalBaseUrl(pluginId: string): Promise<string>;
+}
 
 // @public (undocumented)
 export type StatusCheck = () => Promise<any>;
 
 // @public
-export function statusCheckHandler(options?: StatusCheckHandlerOptions): Promise<RequestHandler>;
+export function statusCheckHandler(
+  options?: StatusCheckHandlerOptions,
+): Promise<RequestHandler>;
 
 // @public (undocumented)
 export interface StatusCheckHandlerOptions {
-    statusCheck?: StatusCheck;
+  statusCheck?: StatusCheck;
 }
 
 // @public
 export type UrlReader = {
-    read(url: string): Promise<Buffer>;
-    readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
-    search(url: string, options?: SearchOptions): Promise<SearchResponse>;
+  read(url: string): Promise<Buffer>;
+  readTree(url: string, options?: ReadTreeOptions): Promise<ReadTreeResponse>;
+  search(url: string, options?: SearchOptions): Promise<SearchResponse>;
 };
 
 // @public
 export class UrlReaders {
-    static create({ logger, config, factories }: CreateOptions): UrlReader;
-    static default({ logger, config, factories }: CreateOptions): UrlReader;
+  static create({ logger, config, factories }: CreateOptions): UrlReader;
+  static default({ logger, config, factories }: CreateOptions): UrlReader;
 }
 
 // @public
-export function useHotCleanup(_module: NodeModule, cancelEffect: () => void): void;
+export function useHotCleanup(
+  _module: NodeModule,
+  cancelEffect: () => void,
+): void;
 
 // @public
 export function useHotMemoize<T>(_module: NodeModule, valueFactory: () => T): T;
 
-
 // (No @packageDocumentation comment for this package)
-
 ```
diff --git a/packages/backend-common/config.d.ts b/packages/backend-common/config.d.ts
index 875660a594..d735f39935 100644
--- a/packages/backend-common/config.d.ts
+++ b/packages/backend-common/config.d.ts
@@ -14,6 +14,23 @@
  * limitations under the License.
  */
 
+export type PluginDatabaseConfig =
+  | {
+      /** Database client to use for plugin. */
+      client?: 'sqlite3';
+      /** Database connection to use with plugin. */
+      connection?: ':memory:' | string | { filename: string };
+    }
+  | {
+      /** Database client to use for plugin. */
+      client?: 'pg';
+      /**
+       * PostgreSQL connection string or knex configuration object for plugin.
+       * @secret
+       */
+      connection?: string | object;
+    };
+
 export interface Config {
   app: {
     baseUrl: string; // defined in core, but repeated here without doc
@@ -58,6 +75,12 @@ export interface Config {
       | {
           client: 'sqlite3';
           connection: ':memory:' | string | { filename: string };
+          /** Optional sqlite3 database filename prefix. */
+          prefix?: string;
+          /** Override database config per plugin.  */
+          plugin?: {
+            [pluginId: string]: PluginDatabaseConfig;
+          };
         }
       | {
           client: 'pg';
@@ -66,6 +89,12 @@ export interface Config {
            * @secret
            */
           connection: string | object;
+          /** Optional PostgreSQL database prefix. */
+          prefix?: string;
+          /** Override database config per plugin.  */
+          plugin?: {
+            [pluginId: string]: PluginDatabaseConfig;
+          };
         };
 
     /** Cache connection configuration, select cache type using the `store` field */
diff --git a/packages/backend-common/src/database/PluginConnection.test.ts b/packages/backend-common/src/database/PluginConnection.test.ts
new file mode 100644
index 0000000000..e3a5157d7d
--- /dev/null
+++ b/packages/backend-common/src/database/PluginConnection.test.ts
@@ -0,0 +1,325 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { ConfigReader } from '@backstage/config';
+import { omit } from 'lodash';
+import { createDatabaseClient, ensureDatabaseExists } from './connection';
+import { PluginConnectionDatabaseManager } from './PluginConnection';
+
+jest.mock('./connection', () => ({
+  ...jest.requireActual('./connection'),
+  createDatabaseClient: jest.fn(),
+  ensureDatabaseExists: jest.fn(),
+}));
+
+describe('PluginConnectionDatabaseManager', () => {
+  // This is similar to the ts-jest `mocked` helper.
+  const mocked = (f: Function) => f as jest.Mock;
+
+  afterEach(() => jest.resetAllMocks());
+
+  describe('PluginConnectionDatabaseManager.fromConfig', () => {
+    const backendConfig = {
+      backend: {
+        database: {
+          client: 'pg',
+          connection: {
+            host: 'localhost',
+            user: 'foo',
+            password: 'bar',
+            database: 'foodb',
+          },
+        },
+      },
+    };
+    const defaultConfig = () => new ConfigReader(backendConfig);
+
+    it('accesses the backend.database key', () => {
+      const getConfig = jest.fn();
+      const config = defaultConfig();
+      config.getConfig = getConfig;
+
+      PluginConnectionDatabaseManager.fromConfig(config);
+
+      expect(getConfig.mock.calls[0][0]).toEqual('backend.database');
+    });
+  });
+
+  describe('PluginConnectionDatabaseManager.forPlugin', () => {
+    const config = {
+      backend: {
+        database: {
+          client: 'pg',
+          prefix: 'test_prefix_',
+          connection: {
+            host: 'localhost',
+            user: 'foo',
+            password: 'bar',
+            database: 'foodb',
+          },
+          plugin: {
+            testdbname: {
+              connection: {
+                database: 'database_name_overriden',
+              },
+            },
+            differentclient: {
+              client: 'sqlite3',
+              connection: {
+                filename: 'plugin_with_different_client',
+              },
+            },
+            differentclientconnstring: {
+              client: 'sqlite3',
+              connection: ':inmemory:',
+            },
+            stringoverride: {
+              connection: 'postgresql://testuser:testpass@acme:5432/userdbname',
+            },
+          },
+        },
+      },
+    };
+    const manager = PluginConnectionDatabaseManager.fromConfig(
+      new ConfigReader(config),
+    );
+
+    it('connects to a plugin database using default config', async () => {
+      const pluginId = 'pluginwithoutconfig';
+
+      await manager.forPlugin(pluginId).getClient();
+      expect(mocked(createDatabaseClient)).toHaveBeenCalledTimes(1);
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, overrides] = mockCalls[0];
+
+      // default config should be passed through to underlying connector
+      expect(baseConfig.get()).toMatchObject({
+        client: 'pg',
+        connection: omit(config.backend.database.connection, ['database']),
+      });
+
+      // override using database name generated from pluginId and prefix
+      expect(overrides).toMatchObject({
+        connection: {
+          database: `${config.backend.database.prefix}${pluginId}`,
+        },
+      });
+    });
+
+    it('provides a plugin db which uses components from top level connection string', async () => {
+      const testManager = PluginConnectionDatabaseManager.fromConfig(
+        new ConfigReader({
+          backend: {
+            database: {
+              client: 'pg',
+              connection: 'postgresql://foo:bar@acme:5432/foodb',
+            },
+          },
+        }),
+      );
+
+      await testManager.forPlugin('pluginwithoutconfig').getClient();
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, overrides] = mockCalls[0];
+
+      // parsed connection string **without** db name should be passed through
+      expect(baseConfig.get()).toMatchObject({
+        connection: {
+          host: 'acme',
+          user: 'foo',
+          password: 'bar',
+          port: '5432',
+        },
+      });
+
+      // we expect a pg database name override with ${prefix} followed by pluginId
+      expect(overrides).toHaveProperty(
+        'connection.database',
+        expect.stringContaining('pluginwithoutconfig'),
+      );
+    });
+
+    it('uses top level sqlite database filename if plugin config is not present', async () => {
+      const testManager = PluginConnectionDatabaseManager.fromConfig(
+        new ConfigReader({
+          backend: {
+            database: {
+              client: 'sqlite3',
+              connection: 'some-file-path',
+            },
+          },
+        }),
+      );
+
+      await testManager.forPlugin('pluginwithoutconfig').getClient();
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [_, overrides] = mockCalls[0];
+
+      expect(overrides).toHaveProperty(
+        'connection.filename',
+        expect.stringContaining('some-file-path'),
+      );
+    });
+
+    it('provides an inmemory sqlite database if top level is also inmemory and plugin config is not present', async () => {
+      const testManager = PluginConnectionDatabaseManager.fromConfig(
+        new ConfigReader({
+          backend: {
+            database: {
+              client: 'sqlite3',
+              connection: ':inmemory:',
+            },
+          },
+        }),
+      );
+
+      await testManager.forPlugin('pluginwithoutconfig').getClient();
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [_, overrides] = mockCalls[0];
+
+      expect(overrides).toHaveProperty(
+        'connection.filename',
+        expect.stringContaining(':inmemory:'),
+      );
+    });
+
+    it('connects to a plugin database using a specific database name', async () => {
+      // testdbname.connection.database is set in config
+      await manager.forPlugin('testdbname').getClient();
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [_baseConfig, overrides] = mockCalls[0];
+
+      // simple case where only database name is overriden
+      expect(overrides).toMatchObject({
+        connection: {
+          database: 'database_name_overriden',
+        },
+      });
+    });
+
+    it('ensure plugin specific database is created', async () => {
+      const pluginId = 'testdbname';
+      // testdbname.connection.database is set in config
+      await manager.forPlugin(pluginId).getClient();
+
+      const mockCalls = mocked(ensureDatabaseExists).mock.calls.splice(-1);
+      const [_, dbname] = mockCalls[0];
+
+      expect(dbname).toEqual(
+        config.backend.database.plugin[pluginId].connection.database,
+      );
+    });
+
+    it('provides different plugins with their own databases', async () => {
+      await manager.forPlugin('plugin1').getClient();
+      await manager.forPlugin('plugin2').getClient();
+
+      expect(mocked(createDatabaseClient)).toHaveBeenCalledTimes(2);
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls;
+      const [plugin1CallArgs, plugin2CallArgs] = mockCalls;
+
+      // database name overrides should be different
+      expect(plugin1CallArgs[1].connection.database).not.toEqual(
+        plugin2CallArgs[1].connection.database,
+      );
+    });
+
+    it('uses plugin connection as base if default client is different from plugin client', async () => {
+      const pluginId = 'differentclient';
+      await manager.forPlugin(pluginId).getClient();
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, _overrides] = mockCalls[0];
+
+      // plugin connection should be used as base config, client is different
+      expect(baseConfig.get()).toMatchObject({
+        client: 'sqlite3',
+        connection: config.backend.database.plugin[pluginId].connection,
+      });
+    });
+
+    it('provides database client specific base and override when client set under plugin', async () => {
+      const pluginId = 'differentclient';
+      await manager.forPlugin(pluginId).getClient();
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, overrides] = mockCalls[0];
+
+      // plugin client should be sqlite3
+      expect(baseConfig.get().client).toEqual('sqlite3');
+
+      // sqlite3 uses 'filename' instead of 'database'
+      expect(overrides).toHaveProperty('connection.filename');
+    });
+
+    it('provides database client specific base from plugin connection string when client set under plugin', async () => {
+      const pluginId = 'differentclientconnstring';
+      await manager.forPlugin(pluginId).getClient();
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, overrides] = mockCalls[0];
+
+      expect(baseConfig.get().client).toEqual('sqlite3');
+
+      expect(overrides).toHaveProperty('connection.filename', ':inmemory:');
+    });
+
+    it('generates a database name override when prefix is not explicitly set', async () => {
+      const testManager = PluginConnectionDatabaseManager.fromConfig(
+        new ConfigReader({
+          backend: {
+            database: {
+              client: 'pg',
+              connection: {
+                host: 'localhost',
+                user: 'foo',
+                password: 'bar',
+                database: 'foodb',
+              },
+            },
+          },
+        }),
+      );
+
+      await testManager.forPlugin('testplugin').getClient();
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [_baseConfig, overrides] = mockCalls[0];
+
+      expect(overrides).toHaveProperty(
+        'connection.database',
+        expect.stringContaining(PluginConnectionDatabaseManager.DEFAULT_PREFIX),
+      );
+    });
+
+    it('uses values from plugin connection string if top level client should be used', async () => {
+      const pluginId = 'stringoverride';
+      await manager.forPlugin(pluginId).getClient();
+
+      const mockCalls = mocked(createDatabaseClient).mock.calls.splice(-1);
+      const [baseConfig, overrides] = mockCalls[0];
+
+      // plugin client should be pg
+      expect(baseConfig.get().client).toEqual('pg');
+
+      expect(overrides).toHaveProperty(
+        'connection.database',
+        expect.stringContaining('userdbname'),
+      );
+    });
+  });
+});
diff --git a/packages/backend-common/src/database/PluginConnection.ts b/packages/backend-common/src/database/PluginConnection.ts
new file mode 100644
index 0000000000..18c209c8d5
--- /dev/null
+++ b/packages/backend-common/src/database/PluginConnection.ts
@@ -0,0 +1,160 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { Knex } from 'knex';
+import { omit } from 'lodash';
+import { Config, ConfigReader } from '@backstage/config';
+import {
+  createDatabaseClient,
+  ensureDatabaseExists,
+  createNameOverride,
+  normalizeConnection,
+} from './connection';
+import { PluginDatabaseManager } from './types';
+
+function pluginPath(pluginId: string): string {
+  return `plugin.${pluginId}`;
+}
+
+export class PluginConnectionDatabaseManager {
+  static readonly DEFAULT_PREFIX = 'backstage_plugin_';
+
+  /**
+   * Creates a PluginConnectionDatabaseManager from `backend.database` config.
+   *
+   * The database manager allows the user to set connection and client settings on a per pluginId
+   * basis by defining a database config block under `plugin.<pluginId>` in addition to top level
+   * defaults. Optionally, a user may set `prefix` which is used to prefix generated database
+   * names if config is not provided.
+   *
+   * @param config The loaded application configuration.
+   */
+  static fromConfig(config: Config): PluginConnectionDatabaseManager {
+    return new PluginConnectionDatabaseManager(
+      config.getConfig('backend.database'),
+    );
+  }
+
+  private constructor(private readonly config: Config) {}
+
+  /**
+   * Generates a PluginDatabaseManager for consumption by plugins.
+   *
+   * @param pluginId The plugin that the database manager should be created for. Plugin names should be unique
+   * as they are used to look up database config overrides under `backend.database.plugin`.
+   */
+  forPlugin(pluginId: string): PluginDatabaseManager {
+    const _this = this;
+
+    return {
+      getClient(): Promise<Knex> {
+        return _this.getDatabase(pluginId);
+      },
+    };
+  }
+
+  /**
+   * Provides the canonical database name for a given pluginId.
+   *
+   * This method provides the effective database name which is determined using global
+   * and plugin specific database config. If no explicit database name is configured,
+   * this method will provide a generated name which is the pluginId prefixed using
+   * the value from `PluginConnectionDatabaseManager.DEFAULT_PREFIX`.
+   *
+   * @param pluginId Lookup the database name for given plugin
+   * */
+  getDatabaseName(pluginId: string): string {
+    const pluginConfig: Config = this.getConfigForPlugin(pluginId);
+
+    // determine root sqlite config to pass through as this is a special case
+    const rootConnection = this.config.get('connection');
+    const rootSqliteName =
+      typeof rootConnection === 'string'
+        ? rootConnection
+        : this.config.getOptionalString('connection.filename') ?? ':inmemory:';
+
+    const prefix =
+      this.config.getOptionalString('prefix') ??
+      PluginConnectionDatabaseManager.DEFAULT_PREFIX;
+
+    const isSqlite = this.config.getString('client') === 'sqlite3';
+    return (
+      // attempt to lookup pg and mysql database name
+      pluginConfig.getOptionalString('connection.database') ??
+      // attempt to lookup sqlite3 database file name
+      pluginConfig.getOptionalString('connection.filename') ??
+      // if root is sqlite - attempt to use top level connection, fallback to :inmemory:
+      (isSqlite ? rootSqliteName : null) ??
+      // generate a database name using prefix and pluginId
+      `${prefix}${pluginId}`
+    );
+  }
+
+  /**
+   * Provides a base database connector config by merging different config sources.
+   *
+   * This method provides a baseConfig for a database connector without the target
+   * database's name property ('database', 'filename'). The client type is determined
+   * by plugin specific config which uses the default as the fallback.
+   *
+   * If the client type is the same as the plugin or not specified, the global
+   * connection config will be extended with plugin specific config.
+   *
+   * @param pluginId The plugin that the database baseConfig should correspond to
+   * */
+  private getConfigForPlugin(pluginId: string): Config {
+    const pluginConfig = this.config.getOptionalConfig(pluginPath(pluginId));
+
+    const baseClient = this.config.getString('client');
+    const client = pluginConfig?.getOptionalString('client') ?? baseClient;
+
+    const baseConnection = normalizeConnection(
+      this.config.get('connection'),
+      baseClient,
+    );
+    const connection = normalizeConnection(
+      pluginConfig?.getOptional('connection') ?? {},
+      client,
+    );
+
+    return new ConfigReader({
+      client,
+      connection: {
+        // if same client type, extend original connection config without dbname config
+        ...(client === baseClient
+          ? omit(baseConnection, ['database', 'filename'])
+          : {}),
+        ...connection,
+      },
+    });
+  }
+
+  private async getDatabase(pluginId: string): Promise<Knex> {
+    const pluginConfig = this.getConfigForPlugin(pluginId);
+
+    await ensureDatabaseExists(pluginConfig, this.getDatabaseName(pluginId));
+    return createDatabaseClient(
+      pluginConfig,
+      this.getDatabaseOverrides(pluginId),
+    );
+  }
+
+  private getDatabaseOverrides(pluginId: string): Knex.Config {
+    return createNameOverride(
+      this.getConfigForPlugin(pluginId).get('client'),
+      this.getDatabaseName(pluginId),
+    );
+  }
+}
diff --git a/packages/backend-common/src/database/connection.test.ts b/packages/backend-common/src/database/connection.test.ts
index 6ab163ff7f..fa7ccd480a 100644
--- a/packages/backend-common/src/database/connection.test.ts
+++ b/packages/backend-common/src/database/connection.test.ts
@@ -15,7 +15,11 @@
  */
 
 import { ConfigReader } from '@backstage/config';
-import { createDatabaseClient } from './connection';
+import {
+  createDatabaseClient,
+  createNameOverride,
+  parseConnectionString,
+} from './connection';
 
 describe('database connection', () => {
   describe('createDatabaseClient', () => {
@@ -103,4 +107,49 @@ describe('database connection', () => {
       ).toThrowError();
     });
   });
+
+  describe('createNameOverride', () => {
+    it('returns Knex config for postgres', () => {
+      expect(createNameOverride('pg', 'testpg')).toHaveProperty(
+        'connection.database',
+        'testpg',
+      );
+    });
+
+    it('returns Knex config for sqlite', () => {
+      expect(createNameOverride('sqlite3', 'testsqlite')).toHaveProperty(
+        'connection.filename',
+        'testsqlite',
+      );
+    });
+
+    it('returns Knex config for mysql', () => {
+      expect(createNameOverride('mysql', 'testmysql')).toHaveProperty(
+        'connection.database',
+        'testmysql',
+      );
+    });
+
+    it('throws an error for unknown connection', () => {
+      expect(() => createNameOverride('unknown', 'testname')).toThrowError();
+    });
+  });
+
+  describe('parseConnectionString', () => {
+    it('returns parsed Knex.StaticConnectionConfig for postgres', () => {
+      expect(
+        parseConnectionString('postgresql://foo:bar@acme:5432/foodb', 'pg'),
+      ).toHaveProperty('database', 'foodb');
+    });
+
+    it('returns parsed Knex.StaticConnectionConfig for mysql2', () => {
+      expect(
+        parseConnectionString('mysql://foo:bar@acme:3306/foodb', 'mysql2'),
+      ).toHaveProperty('database', 'foodb');
+    });
+
+    it('throws an error if client hint is not provided', () => {
+      expect(() => parseConnectionString('sqlite://')).toThrow();
+    });
+  });
 });
diff --git a/packages/backend-common/src/database/connection.ts b/packages/backend-common/src/database/connection.ts
index fb16915cf4..4a4a8c6ba2 100644
--- a/packages/backend-common/src/database/connection.ts
+++ b/packages/backend-common/src/database/connection.ts
@@ -14,14 +14,30 @@
  * limitations under the License.
  */
 
-import { Config } from '@backstage/config';
+import { Config, JsonObject } from '@backstage/config';
+import { InputError } from '@backstage/errors';
 import knexFactory, { Knex } from 'knex';
 import { mergeDatabaseConfig } from './config';
-import { createMysqlDatabaseClient, ensureMysqlDatabaseExists } from './mysql';
-import { createPgDatabaseClient, ensurePgDatabaseExists } from './postgres';
-import { createSqliteDatabaseClient } from './sqlite3';
+import { DatabaseConnector } from './connector';
 
-type DatabaseClient = 'pg' | 'sqlite3' | string;
+import { mysqlConnector } from './mysql';
+import { pgConnector } from './postgres';
+import { sqlite3Connector } from './sqlite3';
+
+type DatabaseClient = 'pg' | 'sqlite3' | 'mysql' | 'mysql2' | string;
+
+/**
+ * Mapping of client type to supported database connectors
+ *
+ * Database connectors can be aliased here, for example mysql2 uses
+ * the same connector as mysql.
+ * */
+const ConnectorMapping: Record<DatabaseClient, DatabaseConnector> = {
+  pg: pgConnector,
+  sqlite3: sqlite3Connector,
+  mysql: mysqlConnector,
+  mysql2: mysqlConnector,
+};
 
 /**
  * Creates a knex database connection
@@ -35,15 +51,10 @@ export function createDatabaseClient(
 ) {
   const client: DatabaseClient = dbConfig.getString('client');
 
-  if (client === 'pg') {
-    return createPgDatabaseClient(dbConfig, overrides);
-  } else if (client === 'mysql' || client === 'mysql2') {
-    return createMysqlDatabaseClient(dbConfig, overrides);
-  } else if (client === 'sqlite3') {
-    return createSqliteDatabaseClient(dbConfig, overrides);
-  }
-
-  return knexFactory(mergeDatabaseConfig(dbConfig.get(), overrides));
+  return (
+    ConnectorMapping[client]?.createClient(dbConfig, overrides) ??
+    knexFactory(mergeDatabaseConfig(dbConfig.get(), overrides))
+  );
 }
 
 /**
@@ -61,11 +72,63 @@ export async function ensureDatabaseExists(
 ) {
   const client: DatabaseClient = dbConfig.getString('client');
 
-  if (client === 'pg') {
-    return ensurePgDatabaseExists(dbConfig, ...databases);
-  } else if (client === 'mysql' || client === 'mysql2') {
-    return ensureMysqlDatabaseExists(dbConfig, ...databases);
+  return ConnectorMapping[client]?.ensureDatabaseExists?.(
+    dbConfig,
+    ...databases,
+  );
+}
+
+/**
+ * Provides a Knex.Config object with the provided database name for a given client.
+ * */
+export function createNameOverride(
+  client: string,
+  name: string,
+): Partial<Knex.Config> {
+  try {
+    return ConnectorMapping[client].createNameOverride(name);
+  } catch (e) {
+    throw new InputError(
+      `Unable to create database name override for '${client}' connector`,
+      e,
+    );
+  }
+}
+
+/**
+ * Parses a connection string for a given client and provides a connection config.
+ * */
+export function parseConnectionString(
+  connectionString: string,
+  client?: string,
+): Knex.StaticConnectionConfig {
+  if (typeof client === 'undefined' || client === null) {
+    throw new InputError(
+      'Database connection string client type auto-detection is not yet supported.',
+    );
   }
 
-  return undefined;
+  try {
+    return ConnectorMapping[client].parseConnectionString(connectionString);
+  } catch (e) {
+    throw new InputError(
+      `Unable to parse connection string for '${client}' connector`,
+    );
+  }
+}
+
+/**
+ * Normalizes a connection config or string into an object which can be passed to Knex.
+ * */
+export function normalizeConnection(
+  connection: Knex.StaticConnectionConfig | JsonObject | string,
+  client: string,
+): Record<string, any> {
+  if (typeof connection === 'undefined' || connection === null) {
+    return {};
+  }
+
+  return typeof connection === 'string' || connection instanceof String
+    ? parseConnectionString(connection as string, client)
+    : connection;
 }
diff --git a/packages/backend-common/src/database/connector.ts b/packages/backend-common/src/database/connector.ts
new file mode 100644
index 0000000000..0336510f19
--- /dev/null
+++ b/packages/backend-common/src/database/connector.ts
@@ -0,0 +1,30 @@
+/*
+ * Copyright 2021 Spotify AB
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+import { Config } from '@backstage/config';
+import { Knex } from 'knex';
+
+export interface DatabaseConnector {
+  createClient(dbConfig: Config, overrides?: Partial<Knex.Config>): Knex;
+  createNameOverride(name: string): Partial<Knex.Config>;
+  parseConnectionString(
+    connectionString: string,
+    client?: string,
+  ): Knex.StaticConnectionConfig;
+  ensureDatabaseExists?(
+    dbConfig: Config,
+    ...databases: Array<string>
+  ): Promise<void>;
+}
diff --git a/packages/backend-common/src/database/index.ts b/packages/backend-common/src/database/index.ts
index c81153aa62..7a8a81e187 100644
--- a/packages/backend-common/src/database/index.ts
+++ b/packages/backend-common/src/database/index.ts
@@ -17,3 +17,4 @@
 export * from './connection';
 export * from './types';
 export * from './SingleConnection';
+export * from './PluginConnection';
diff --git a/packages/backend-common/src/database/mysql.ts b/packages/backend-common/src/database/mysql.ts
index be4632baf6..3bd874de11 100644
--- a/packages/backend-common/src/database/mysql.ts
+++ b/packages/backend-common/src/database/mysql.ts
@@ -18,6 +18,7 @@ import { Config } from '@backstage/config';
 import { InputError } from '@backstage/errors';
 import knexFactory, { Knex } from 'knex';
 import { mergeDatabaseConfig } from './config';
+import { DatabaseConnector } from './connector';
 import yn from 'yn';
 
 /**
@@ -159,3 +160,23 @@ export async function ensureMysqlDatabaseExists(
     await admin.destroy();
   }
 }
+
+export function createMysqlNameOverride(name: string): Partial<Knex.Config> {
+  return {
+    connection: {
+      database: name,
+    },
+  };
+}
+
+/**
+ * MySql database connector.
+ *
+ * Exposes database connector functionality via an immutable object.
+ * */
+export const mysqlConnector: DatabaseConnector = Object.freeze({
+  createClient: createMysqlDatabaseClient,
+  ensureDatabaseExists: ensureMysqlDatabaseExists,
+  createNameOverride: createMysqlNameOverride,
+  parseConnectionString: parseMysqlConnectionString,
+});
diff --git a/packages/backend-common/src/database/postgres.ts b/packages/backend-common/src/database/postgres.ts
index e05eab86e5..40c6a41504 100644
--- a/packages/backend-common/src/database/postgres.ts
+++ b/packages/backend-common/src/database/postgres.ts
@@ -17,6 +17,7 @@
 import knexFactory, { Knex } from 'knex';
 import { Config } from '@backstage/config';
 import { mergeDatabaseConfig } from './config';
+import { DatabaseConnector } from './connector';
 
 /**
  * Creates a knex postgres database connection
@@ -131,3 +132,23 @@ export async function ensurePgDatabaseExists(
     await admin.destroy();
   }
 }
+
+export function createPgNameOverride(name: string): Partial<Knex.Config> {
+  return {
+    connection: {
+      database: name,
+    },
+  };
+}
+
+/**
+ * PostgreSQL database connector.
+ *
+ * Exposes database connector functionality via an immutable object.
+ * */
+export const pgConnector: DatabaseConnector = Object.freeze({
+  createClient: createPgDatabaseClient,
+  ensureDatabaseExists: ensurePgDatabaseExists,
+  createNameOverride: createPgNameOverride,
+  parseConnectionString: parsePgConnectionString,
+});
diff --git a/packages/backend-common/src/database/sqlite3.ts b/packages/backend-common/src/database/sqlite3.ts
index d4169e3899..98c39b9c97 100644
--- a/packages/backend-common/src/database/sqlite3.ts
+++ b/packages/backend-common/src/database/sqlite3.ts
@@ -19,6 +19,7 @@ import { ensureDirSync } from 'fs-extra';
 import knexFactory, { Knex } from 'knex';
 import path from 'path';
 import { mergeDatabaseConfig } from './config';
+import { DatabaseConnector } from './connector';
 
 /**
  * Creates a knex sqlite3 database connection
@@ -99,3 +100,28 @@ export function buildSqliteDatabaseConfig(
 
   return config;
 }
+
+export function createSqliteNameOverride(name: string): Partial<Knex.Config> {
+  return {
+    connection: parseSqliteConnectionString(name),
+  };
+}
+
+export function parseSqliteConnectionString(
+  name: string,
+): Knex.Sqlite3ConnectionConfig {
+  return {
+    filename: name,
+  };
+}
+
+/**
+ * Sqlite3 database connector.
+ *
+ * Exposes database connector functionality via an immutable object.
+ * */
+export const sqlite3Connector: DatabaseConnector = Object.freeze({
+  createClient: createSqliteDatabaseClient,
+  createNameOverride: createSqliteNameOverride,
+  parseConnectionString: parseSqliteConnectionString,
+});
diff --git a/packages/backend/src/index.ts b/packages/backend/src/index.ts
index d43c8f0101..9cab64e797 100644
--- a/packages/backend/src/index.ts
+++ b/packages/backend/src/index.ts
@@ -29,7 +29,7 @@ import {
   getRootLogger,
   loadBackendConfig,
   notFoundHandler,
-  SingleConnectionDatabaseManager,
+  PluginConnectionDatabaseManager,
   SingleHostDiscovery,
   UrlReaders,
   useHotMemoize,
@@ -59,7 +59,7 @@ function makeCreateEnv(config: Config) {
 
   root.info(`Created UrlReader ${reader}`);
 
-  const databaseManager = SingleConnectionDatabaseManager.fromConfig(config);
+  const databaseManager = PluginConnectionDatabaseManager.fromConfig(config);
   const cacheManager = CacheManager.fromConfig(config);
 
   return (plugin: string): PluginEnvironment => {